Freeze mixed-file retrieval diagnostic and shared-engine evaluator

docs/MIXED-QUALITY.md

@@ -0,0 +1,32 @@
+# Mixed retrieval quality gate
+
+A frozen mixed diagnostic is prepared in [fixtures/mixed](../fixtures/mixed/README.md).
+It addresses the missing comparison between natural-language All-files queries
+and explicit Code scope while images, music, speech, videos and actual documents
+compete. It contains 32 real files and 91 English/German queries, including two
+pinned upstream llama.cpp implementations. The existing media labels are reused
+with their licenses and attribution. They are not an untouched held-out set.
+
+**No actual mixed inference results are recorded yet.** The frozen manifest
+SHA-256 is `f83d27227582f614159ff153654d5bb5a47bf74998141e389557f3950ea0b924`.
+Asset/hash verification and six meaningful scoring/provenance tests passed.
+These prove harness behavior only; they establish no retrieval accuracy.
+
+The [evaluator](../tools/mixed-eval/README.md) calls the shared production engine
+through CLI indexing and MCP search. It compares lexical/hybrid retrieval and
+All-files/Code scope without changing ranking or tuning a cutoff against these
+labels. File-level recall and no-answer returned-result rates remain separate
+from spoken timestamp localization. Raw evidence must preserve actual failures,
+encoding errors and unsupported filters.
+
+Remaining work:
+
+- Run the frozen corpus with the verified Google model and CPU runtime in a
+  coordinated slot; record exact CLI/worker hashes and actual raw results.
+- Inspect failures by modality, language, requested artifact type and scope;
+  distinguish correct-file retrieval from a useful passage/timestamp.
+- Author a larger independently reviewed held-out corpus before ranking changes;
+  frozen diagnostics must not become the sole optimization target.
+- Establish recall and no-answer acceptance targets on representative user
+  libraries, unrelated repositories and longer recordings. Tiny diagnostics
+  cannot close the production relevance gate.

fixtures/mixed/LLAMA-LICENSE.txt

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023-2026 The ggml authors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

fixtures/mixed/README.md

@@ -0,0 +1,34 @@
+# Mixed All-files diagnostic v1
+
+This is a frozen, small diagnostic, **not a production quality score**. No mixed
+model run has been performed yet.
+
+The corpus contains 32 actual files: six searchi implementations, four searchi
+technical documents, two complete pinned llama.cpp source files, and twenty
+existing licensed media files. Source snapshots retain exact original bytes;
+`fixture.json` records upstream paths, commits, SHA-256, sizes, and licenses.
+The llama.cpp copies come from committed upstream objects, excluding local
+runtime patches. Its MIT notice is in `LLAMA-LICENSE.txt`. Media files are
+referenced from `../media/assets` rather than duplicated. Their source URLs,
+licenses, independently observed labels and attribution remain in
+[the media fixture](../media/README.md).
+
+There are 91 frozen queries: English/German implementation and document pairs,
+65 reused media queries, four additional absent-content queries and four pinned
+upstream implementation queries. Code queries run in both All files and Code
+scope. Documents about the implementation coexist with the code, providing
+realistic competing content. Speech, music, images and videos compete with
+both. The case asking for code that chooses speech search is intentional: it
+can expose audio-intent routing that suppresses a relevant implementation.
+
+Labels were written by one annotator before **mixed** inference. Existing media
+results had already been observed in their separate evaluation, so this is not
+an untouched held-out media benchmark. searchi sources are the application's
+own repository; the two llama.cpp files add external code, but do not establish
+quality across unrelated real projects. Never adjust v1 labels after seeing
+mixed results. Corrections require a new fixture version, explicit reasoning,
+and a new pre-inference lock. A second blinded human review remains open.
+
+Run `python tools/mixed-eval/evaluate.py --verify-only` to verify labels and all
+source/media bytes without invoking a model. See
+[the evaluator instructions](../../tools/mixed-eval/README.md) for real runs.

fixtures/mixed/labels.lock.json

@@ -0,0 +1,5 @@
+{
+  "version": 1,
+  "fixture_sha256": "f83d27227582f614159ff153654d5bb5a47bf74998141e389557f3950ea0b924",
+  "model_runs": 0
+}

fixtures/mixed/sources/apps__desktop__src__search-performance.ts

@@ -0,0 +1,230 @@
+/** UI times use performance.now(); native spans use their own command clock. */
+export interface NativeSpan {
+  name: string;
+  start_ms: number;
+  duration_ms: number;
+  parent: number | null;
+  counters: Record<string, number>;
+}
+export interface NativeMetrics {
+  total_ms: number;
+  result_count: number;
+  trace: NativeSpan[] | null;
+  dropped_spans: number;
+}
+export interface MeasuredResponse<T> {
+  results: T[];
+  metrics: NativeMetrics;
+}
+export type SearchLane = "lexical" | "hybrid";
+export interface AttemptStart {
+  lane: SearchLane;
+  startedAt: number;
+  submittedAt: number;
+  submission: number;
+}
+interface Attempt {
+  lane: SearchLane;
+  submission: number;
+  attempt: number;
+  queued_ms: number;
+  retry_wait_ms: number;
+  invoke_ms: number;
+  native: NativeMetrics | null;
+  transport_estimate_ms: number | null;
+  failed: boolean;
+  ended_at: number;
+}
+interface InputTimingSummary {
+  debounce_ms: number;
+  first_results_ms: number | null;
+  latest_displayed_ms: number | null;
+  hybrid_final_results_ms: number | null;
+}
+
+export class SearchMeasurement {
+  private originalInput: Readonly<InputTimingSummary> | null = null;
+  debug: boolean;
+  inputAt: number;
+  scheduledAt: number;
+  context: Record<string, string | number | boolean | null>;
+  submissions = new Map<SearchLane, number>();
+  latestSubmission = new Map<SearchLane, number>();
+  submissionSequence = new Map<SearchLane, number>();
+  laneAttempts = new Map<SearchLane, number>();
+  attempts: Attempt[] = [];
+  droppedAttempts = 0;
+  droppedDisplays = 0;
+  firstDisplayMs: number | null = null;
+  displays: {
+    lane: SearchLane;
+    elapsed_ms: number;
+    dom_update_ms: number;
+    count: number;
+  }[] = [];
+
+  constructor(
+    inputAt: number,
+    scheduledAt: number,
+    debug: boolean,
+    context: Record<string, string | number | boolean | null> = {},
+  ) {
+    this.inputAt = inputAt;
+    this.scheduledAt = scheduledAt;
+    this.debug = debug;
+    this.context = { ...context };
+  }
+  /** Filesystem updates are independent requests, not elapsed user idle time. */
+  freshForRefresh(now: number): SearchMeasurement {
+    const refreshed = new SearchMeasurement(now, now, this.debug, this.context);
+    // Preserve one fixed-size numeric snapshot, never an ancestor trace/history.
+    // Pending responses on the old request cannot rewrite this snapshot.
+    refreshed.originalInput = Object.freeze({
+      ...(this.originalInput ?? {
+        debounce_ms: Math.max(0, this.scheduledAt - this.inputAt),
+        first_results_ms: this.firstDisplayMs,
+        latest_displayed_ms: this.displays.at(-1)?.elapsed_ms ?? null,
+        hybrid_final_results_ms:
+          [...this.displays]
+            .reverse()
+            .find((display) => display.lane === "hybrid")?.elapsed_ms ?? null,
+      }),
+    });
+    return refreshed;
+  }
+  submitted(lane: SearchLane, at: number) {
+    if (!this.submissions.has(lane)) this.submissions.set(lane, at);
+    this.latestSubmission.set(lane, at);
+    this.submissionSequence.set(
+      lane,
+      (this.submissionSequence.get(lane) ?? 0) + 1,
+    );
+  }
+  begin(lane: SearchLane, startedAt: number): AttemptStart {
+    return {
+      lane,
+      startedAt,
+      submittedAt: this.latestSubmission.get(lane) ?? startedAt,
+      submission: this.submissionSequence.get(lane) ?? 0,
+    };
+  }
+  completed(
+    token: AttemptStart,
+    end: number,
+    native: NativeMetrics | null,
+    failed: boolean,
+  ) {
+    const { lane, startedAt: start, submittedAt, submission } = token;
+    const prior = this.attempts.filter((attempt) => attempt.lane === lane);
+    const previous = prior.at(-1);
+    const retry = previous?.failed && previous.submission === submission;
+    const attemptNumber = (this.laneAttempts.get(lane) ?? 0) + 1;
+    this.laneAttempts.set(lane, attemptNumber);
+    const elapsed = Math.max(0, end - start);
+    this.attempts.push({
+      lane,
+      submission,
+      attempt: attemptNumber,
+      queued_ms: retry ? 0 : Math.max(0, start - submittedAt),
+      retry_wait_ms: retry ? Math.max(0, start - previous.ended_at) : 0,
+      invoke_ms: elapsed,
+      native,
+      transport_estimate_ms: native
+        ? Math.max(0, elapsed - native.total_ms)
+        : null,
+      failed,
+      ended_at: end,
+    });
+    if (this.attempts.length > (this.debug ? 32 : 4)) {
+      this.attempts.shift();
+      this.droppedAttempts++;
+    }
+  }
+  displayed(lane: SearchLane, start: number, end: number, count: number) {
+    if (this.firstDisplayMs === null)
+      this.firstDisplayMs = Math.max(0, end - this.inputAt);
+    this.displays.push({
+      lane,
+      elapsed_ms: Math.max(0, end - this.inputAt),
+      dom_update_ms: Math.max(0, end - start),
+      count,
+    });
+    if (this.displays.length > (this.debug ? 16 : 2)) {
+      this.displays.shift();
+      this.droppedDisplays++;
+    }
+  }
+  latestNative(lane: SearchLane) {
+    return (
+      this.attempts
+        .filter((attempt) => attempt.lane === lane && !attempt.failed)
+        .at(-1)?.native ?? null
+    );
+  }
+  /** Deliberately exclude query text, filenames, paths, snippets and errors. */
+  export() {
+    return {
+      version: 1,
+      trigger: this.originalInput ? "filesystem_refresh" : "user_search",
+      original_input: this.originalInput ? { ...this.originalInput } : null,
+      context: this.context,
+      clock_note:
+        "UI monotonic elapsed times; native command-relative spans. Nested spans and parallel lanes must not be summed.",
+      debounce_ms: Math.max(0, this.scheduledAt - this.inputAt),
+      lane_schedule_ms: Object.fromEntries(
+        [...this.submissions].map(([lane, at]) => [
+          lane,
+          Math.max(0, at - this.scheduledAt),
+        ]),
+      ),
+      attempts: this.attempts.map(({ ended_at: _end, ...attempt }) => attempt),
+      displays: this.displays,
+      first_results_ms: this.firstDisplayMs,
+      latest_displayed_ms: this.displays.at(-1)?.elapsed_ms ?? null,
+      hybrid_final_results_ms:
+        [...this.displays]
+          .reverse()
+          .find((display) => display.lane === "hybrid")?.elapsed_ms ?? null,
+      dropped_attempts: this.droppedAttempts,
+      dropped_displays: this.droppedDisplays,
+    };
+  }
+}
+
+export function measuredFailure(
+  error: unknown,
+): { message: string; metrics: NativeMetrics } | null {
+  if (!error || typeof error !== "object") return null;
+  const value = error as { message?: unknown; metrics?: NativeMetrics };
+  return typeof value.message === "string" &&
+    value.metrics &&
+    Number.isFinite(value.metrics.total_ms)
+    ? { message: value.message, metrics: value.metrics }
+    : null;
+}
+
+/** Malformed/cyclic parent chains cannot trap the debug renderer. */
+export function spanDepth(spans: NativeSpan[], index: number) {
+  let depth = 0;
+  const visited = new Set([index]);
+  let parent = spans[index]?.parent;
+  while (
+    parent != null &&
+    spans[parent] &&
+    !visited.has(parent) &&
+    depth < 16
+  ) {
+    visited.add(parent);
+    depth++;
+    parent = spans[parent].parent;
+  }
+  return depth;
+}
+
+export function briefMs(value: number) {
+  return value < 1
+    ? "<1"
+    : value < 10
+      ? value.toFixed(1)
+      : String(Math.round(value));
+}

fixtures/mixed/sources/apps__desktop__src__selection-state.ts

@@ -0,0 +1,24 @@
+/** Preserve the user's result choice while progressive indexing hides it briefly. */
+export class ResultSelection {
+  private scope: string | null = null;
+  private intendedPath: string | null = null;
+
+  begin(scope: string) {
+    if (scope !== this.scope) {
+      this.scope = scope;
+      this.intendedPath = null;
+    }
+  }
+
+  choose(path: string) {
+    this.intendedPath = path;
+  }
+
+  resolve(paths: readonly string[]): number {
+    if (!paths.length) return -1;
+    if (this.intendedPath === null) this.intendedPath = paths[0];
+    const intended = paths.indexOf(this.intendedPath);
+    // A temporary fallback is presentation, not a new user choice.
+    return intended >= 0 ? intended : 0;
+  }
+}

fixtures/mixed/sources/crates__searchi-core__src__extraction.rs

@@ -0,0 +1,320 @@
+//! 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 sha2::{Digest, Sha256};
+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) struct SourceInput {
+    bytes: Vec<u8>,
+    pub hash: String,
+}
+
+/// Read once: the fingerprint and parser consume precisely the same bytes.
+pub(crate) fn read_source(path: &Path) -> Result<Option<SourceInput>> {
+    let extension = path
+        .extension()
+        .and_then(|value| value.to_str())
+        .unwrap_or("")
+        .to_ascii_lowercase();
+    let document = matches!(extension.as_str(), "pdf" | "docx");
+    if !document
+        && !crate::is_code_path(path)
+        && !matches!(
+            extension.as_str(),
+            "txt"
+                | "md"
+                | "json"
+                | "toml"
+                | "yaml"
+                | "yml"
+                | "csv"
+                | "html"
+                | "css"
+                | "xml"
+                | "log"
+                | "sh"
+        )
+    {
+        return Ok(None);
+    }
+    if !document && File::open(path)?.metadata()?.len() > MAX_TEXT as u64 {
+        return Ok(None);
+    }
+    let bytes = read_bounded(path, if document { MAX_DOCUMENT } else { MAX_TEXT })?;
+    let hash = format!("{:x}", Sha256::digest(&bytes));
+    Ok(Some(SourceInput { bytes, hash }))
+}
+
+pub(crate) fn extract(path: &Path, input: SourceInput) -> Result<Option<Vec<Passage>>> {
+    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 code = crate::is_code_path(path);
+    if input.bytes.contains(&0) {
+        return Ok(None);
+    }
+    let Ok(text) = String::from_utf8(input.bytes) else {
+        return Ok(None);
+    };
+    let lines = text.split_inclusive('\n').collect::<Vec<_>>();
+    if code {
+        let mut passages = Vec::new();
+        for start in (0..lines.len()).step_by(96) {
+            if passages.len() >= MAX_PASSAGES {
+                return Ok(None);
+            }
+            let end = (start + 128).min(lines.len());
+            passages.push(Passage {
+                text: lines[start..end].concat(),
+                source: SourceLocation {
+                    kind: "line".into(),
+                    number: (start + 1) as u32,
+                },
+            });
+            if end == lines.len() {
+                break;
+            }
+        }
+        return Ok(Some(passages));
+    }
+    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)
+}
+
+#[cfg(test)]
+mod provenance_tests {
+    use super::*;
+
+    #[test]
+    fn parsing_and_fingerprint_share_one_snapshot_even_when_path_changes() -> Result<()> {
+        let directory = tempfile::tempdir()?;
+        let path = directory.path().join("notes.txt");
+        std::fs::write(&path, "original parser input")?;
+        let input = read_source(&path)?.unwrap();
+        assert_eq!(
+            input.hash,
+            format!("{:x}", Sha256::digest(b"original parser input"))
+        );
+        std::fs::write(&path, "replacement on disk")?;
+        let passages = extract(&path, input)?.unwrap();
+        assert_eq!(passages[0].text, "original parser input");
+        Ok(())
+    }
+}

fixtures/mixed/sources/crates__searchi-service__src__coordination.rs

@@ -0,0 +1,417 @@
+//! Shared admission for persistent inference sessions. This bounds admitted
+//! native worker threads, not all extraction/ANN/OS work. Already executing
+//! inference cannot be preempted; long audio windows remain a latency limit.
+use crate::EmbeddingProvider;
+use anyhow::{bail, Result};
+use searchi_core::metrics::{SpanToken, TraceCollector};
+use searchi_semantic::TextWindow;
+use std::{
+    collections::BTreeMap,
+    path::Path,
+    sync::{Arc, Condvar, Mutex},
+    time::{Duration, Instant},
+};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum InferenceRole {
+    Foreground,
+    Background,
+}
+#[derive(Default)]
+struct State {
+    limit: usize,
+    used: usize,
+    foreground: usize,
+}
+struct Shared {
+    state: Mutex<State>,
+    changed: Condvar,
+}
+/// Clone this handle across query, setup and indexing sessions.
+#[derive(Clone)]
+pub struct InferenceBudget(Arc<Shared>);
+impl InferenceBudget {
+    pub fn new(requested: usize) -> Self {
+        Self(Arc::new(Shared {
+            state: Mutex::new(State {
+                limit: clamp(requested),
+                ..State::default()
+            }),
+            changed: Condvar::new(),
+        }))
+    }
+    pub fn threads(&self) -> usize {
+        self.0.state.lock().unwrap_or_else(|e| e.into_inner()).limit
+    }
+    /// Leave one admitted worker thread for background indexing. On a
+    /// single-thread budget, already executing inference still cannot yield.
+    pub fn worker_threads(&self, role: InferenceRole) -> usize {
+        match role {
+            InferenceRole::Foreground => self.threads().saturating_sub(1).max(1),
+            InferenceRole::Background => 1,
+        }
+    }
+    /// Update admission immediately; already admitted work is not cancelled.
+    /// Recreate sessions whose worker thread count exceeds a reduced limit.
+    pub fn configure(&self, requested: usize) -> usize {
+        let limit = clamp(requested);
+        self.0.state.lock().unwrap_or_else(|e| e.into_inner()).limit = limit;
+        self.0.changed.notify_all();
+        limit
+    }
+    /// Register before waiting for a service mutex. This reserves priority, not
+    /// threads, so the foreground does not deadlock with its indexing session.
+    pub fn pending_foreground(&self) -> PendingForeground {
+        self.0
+            .state
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .foreground += 1;
+        self.0.changed.notify_all();
+        PendingForeground(self.clone())
+    }
+    pub fn acquire(
+        &self,
+        role: InferenceRole,
+        actual_worker_threads: usize,
+    ) -> Result<InferencePermit> {
+        let weight = actual_worker_threads.max(1);
+        let _pending = (role == InferenceRole::Foreground).then(|| self.pending_foreground());
+        let deadline = Instant::now() + Duration::from_secs(30);
+        let mut state = self.0.state.lock().unwrap_or_else(|e| e.into_inner());
+        loop {
+            if weight > state.limit {
+                bail!("Inference session uses {weight} threads but shared budget is {}; recreate the session after changing its thread count", state.limit);
+            }
+            if state.used <= state.limit - weight
+                && (role == InferenceRole::Foreground || state.foreground == 0)
+            {
+                state.used += weight;
+                return Ok(InferencePermit {
+                    budget: self.clone(),
+                    weight,
+                });
+            }
+            let remaining = deadline.saturating_duration_since(Instant::now());
+            if remaining.is_zero() {
+                bail!("Timed out waiting for shared inference budget");
+            }
+            state = self
+                .0
+                .changed
+                .wait_timeout(state, remaining)
+                .unwrap_or_else(|e| e.into_inner())
+                .0;
+        }
+    }
+}
+fn clamp(requested: usize) -> usize {
+    requested
+        .clamp(1, 8)
+        .min(std::thread::available_parallelism().map_or(1, usize::from))
+}
+pub struct PendingForeground(InferenceBudget);
+impl Drop for PendingForeground {
+    fn drop(&mut self) {
+        let mut s = self.0 .0.state.lock().unwrap_or_else(|e| e.into_inner());
+        s.foreground -= 1;
+        self.0 .0.changed.notify_all();
+    }
+}
+pub struct InferencePermit {
+    budget: InferenceBudget,
+    weight: usize,
+}
+impl Drop for InferencePermit {
+    fn drop(&mut self) {
+        let mut s = self
+            .budget
+            .0
+            .state
+            .lock()
+            .unwrap_or_else(|e| e.into_inner());
+        s.used -= self.weight;
+        self.budget.0.changed.notify_all();
+    }
+}
+
+/// Delegates genuine inference unchanged. Background documents yield between
+/// individual encodes even when the caller supplies a larger batch.
+pub struct BudgetedProvider {
+    inner: Box<dyn EmbeddingProvider>,
+    budget: InferenceBudget,
+    role: InferenceRole,
+    threads: usize,
+    query_started: bool,
+}
+impl BudgetedProvider {
+    pub fn new(
+        inner: Box<dyn EmbeddingProvider>,
+        budget: &InferenceBudget,
+        role: InferenceRole,
+        actual_worker_threads: usize,
+    ) -> Self {
+        Self {
+            inner,
+            budget: budget.clone(),
+            role,
+            threads: actual_worker_threads.max(1),
+            query_started: false,
+        }
+    }
+    fn permit(&self) -> Result<InferencePermit> {
+        self.budget.acquire(self.role, self.threads)
+    }
+}
+impl EmbeddingProvider for BudgetedProvider {
+    fn document_batch_limit(&self) -> usize {
+        if self.role == InferenceRole::Background {
+            1
+        } else {
+            self.inner.document_batch_limit()
+        }
+    }
+    fn fingerprint(&self) -> &str {
+        self.inner.fingerprint()
+    }
+    fn dimensions(&self) -> usize {
+        self.inner.dimensions()
+    }
+    fn split_windows(&mut self, text: &str, title: &str) -> Result<Vec<TextWindow>> {
+        let _p = self.permit()?;
+        self.inner.split_windows(text, title)
+    }
+    fn embed_documents(&mut self, texts: &[&str], title: &str) -> Result<Vec<Vec<f32>>> {
+        if self.role == InferenceRole::Background {
+            let mut output = Vec::with_capacity(texts.len());
+            for text in texts {
+                let _p = self.permit()?;
+                output.extend(self.inner.embed_documents(&[*text], title)?);
+            }
+            Ok(output)
+        } else {
+            let _p = self.permit()?;
+            self.inner.embed_documents(texts, title)
+        }
+    }
+    fn embed_query(&mut self, query: &str) -> Result<Vec<f32>> {
+        self.query_started = false;
+        let _p = self.permit()?;
+        self.query_started = true;
+        self.inner.embed_query(query)
+    }
+    fn embed_query_debug(&mut self, query: &str) -> Result<Vec<f32>> {
+        self.query_started = false;
+        let _p = self.permit()?;
+        self.query_started = true;
+        self.inner.embed_query_debug(query)
+    }
+    fn embed_code_query(&mut self, query: &str) -> Result<Vec<f32>> {
+        self.query_started = false;
+        let _p = self.permit()?;
+        self.query_started = true;
+        self.inner.embed_code_query(query)
+    }
+    fn embed_code_query_debug(&mut self, query: &str) -> Result<Vec<f32>> {
+        self.query_started = false;
+        let _p = self.permit()?;
+        self.query_started = true;
+        self.inner.embed_code_query_debug(query)
+    }
+    fn query_diagnostics(&self) -> BTreeMap<String, u64> {
+        if self.query_started {
+            self.inner.query_diagnostics()
+        } else {
+            Default::default()
+        }
+    }
+    fn embed_query_traced(
+        &mut self,
+        query: &str,
+        debug: bool,
+        trace: &mut TraceCollector,
+        parent: Option<SpanToken>,
+    ) -> Result<Vec<f32>> {
+        self.query_started = false;
+        let span = trace.begin("semantic.inference.admission_wait", parent);
+        let permit = self.permit();
+        trace.end(
+            span,
+            if trace.is_debug() {
+                BTreeMap::from([
+                    ("worker_threads".into(), self.threads as u64),
+                    ("failed".into(), u64::from(permit.is_err())),
+                ])
+            } else {
+                Default::default()
+            },
+        );
+        let _permit = permit?;
+        self.query_started = true;
+        self.inner.embed_query_traced(query, debug, trace, parent)
+    }
+    fn embed_code_query_traced(
+        &mut self,
+        query: &str,
+        debug: bool,
+        trace: &mut TraceCollector,
+        parent: Option<SpanToken>,
+    ) -> Result<Vec<f32>> {
+        self.query_started = false;
+        let span = trace.begin("semantic.inference.admission_wait", parent);
+        let permit = self.permit();
+        trace.end(
+            span,
+            if trace.is_debug() {
+                BTreeMap::from([
+                    ("worker_threads".into(), self.threads as u64),
+                    ("failed".into(), u64::from(permit.is_err())),
+                ])
+            } else {
+                Default::default()
+            },
+        );
+        let _permit = permit?;
+        self.query_started = true;
+        self.inner
+            .embed_code_query_traced(query, debug, trace, parent)
+    }
+    fn embed_images(&mut self, paths: &[&Path], budget: u32) -> Result<Vec<Vec<f32>>> {
+        if self.role == InferenceRole::Background {
+            let mut output = Vec::with_capacity(paths.len());
+            for path in paths {
+                let _p = self.permit()?;
+                output.extend(self.inner.embed_images(&[*path], budget)?);
+            }
+            return Ok(output);
+        }
+        let _p = self.permit()?;
+        self.inner.embed_images(paths, budget)
+    }
+    fn embed_audio(&mut self, pcm: &[f32]) -> Result<Vec<f32>> {
+        let _p = self.permit()?;
+        self.inner.embed_audio(pcm)
+    }
+    fn worker_info(&self) -> Option<searchi_semantic::WorkerInfo> {
+        self.inner.worker_info()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn failed_admission_clears_prior_worker_diagnostics_without_calling_worker() {
+        use searchi_core::metrics::MetricsMode;
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        struct ObservedInvocation(Arc<AtomicUsize>);
+        impl EmbeddingProvider for ObservedInvocation {
+            fn fingerprint(&self) -> &str {
+                "scheduler-test-only"
+            }
+            fn dimensions(&self) -> usize {
+                0
+            }
+            fn split_windows(&mut self, _: &str, _: &str) -> Result<Vec<TextWindow>> {
+                unreachable!()
+            }
+            fn embed_documents(&mut self, _: &[&str], _: &str) -> Result<Vec<Vec<f32>>> {
+                unreachable!()
+            }
+            fn embed_query(&mut self, _: &str) -> Result<Vec<f32>> {
+                self.0.fetch_add(1, Ordering::SeqCst);
+                Ok(Vec::new())
+            }
+            fn embed_code_query(&mut self, query: &str) -> Result<Vec<f32>> {
+                self.embed_query(query)
+            }
+            fn query_diagnostics(&self) -> BTreeMap<String, u64> {
+                BTreeMap::from([("prior_worker_call".into(), 1)])
+            }
+        }
+        let budget = InferenceBudget::new(1);
+        // Test worker weight two without requiring particular test host hardware.
+        budget.0.state.lock().unwrap().limit = 2;
+        let calls = Arc::new(AtomicUsize::new(0));
+        let mut provider = BudgetedProvider::new(
+            Box::new(ObservedInvocation(calls.clone())),
+            &budget,
+            InferenceRole::Foreground,
+            2,
+        );
+        provider
+            .embed_query_debug("first admitted invocation")
+            .unwrap();
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+        assert_eq!(provider.query_diagnostics()["prior_worker_call"], 1);
+        assert_eq!(budget.configure(1), 1);
+        let mut trace = TraceCollector::new(MetricsMode::Debug);
+        assert!(provider
+            .embed_query_traced("rejected invocation", true, &mut trace, None)
+            .is_err());
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+        assert!(provider.query_diagnostics().is_empty());
+        let metrics = trace.finish(0);
+        let stages = metrics.trace.unwrap();
+        assert_eq!(stages.len(), 1);
+        assert_eq!(stages[0].name, "semantic.inference.admission_wait");
+        assert_eq!(stages[0].counters["failed"], 1);
+        assert_eq!(stages[0].counters["worker_threads"], 2);
+        assert!(stages[0].duration_ms >= 0.0);
+        assert!(provider.embed_code_query_debug("also rejected").is_err());
+        assert!(provider.query_diagnostics().is_empty());
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+        assert_eq!(budget.0.state.lock().unwrap().foreground, 0);
+        assert_eq!(budget.0.state.lock().unwrap().used, 0);
+    }
+    #[test]
+    fn reserved_query_capacity_admits_foreground_during_background() {
+        let budget = InferenceBudget::new(1);
+        // Deliberately exercise a two-thread scheduler independent of CI hardware.
+        budget.0.state.lock().unwrap().limit = 2;
+        assert_eq!(budget.worker_threads(InferenceRole::Foreground), 1);
+        assert_eq!(budget.worker_threads(InferenceRole::Background), 1);
+        let background = budget.acquire(InferenceRole::Background, 1).unwrap();
+        let foreground = budget.acquire(InferenceRole::Foreground, 1).unwrap();
+        assert_eq!(budget.0.state.lock().unwrap().used, 2);
+        drop(foreground);
+        drop(background);
+    }
+    #[test]
+    fn pending_query_blocks_background_until_released() {
+        let budget = InferenceBudget::new(1);
+        let pending = budget.pending_foreground();
+        let other = budget.clone();
+        let (tx, rx) = std::sync::mpsc::channel();
+        let worker = std::thread::spawn(move || {
+            let _permit = other.acquire(InferenceRole::Background, 1).unwrap();
+            tx.send(()).unwrap();
+        });
+        assert!(rx.recv_timeout(Duration::from_millis(50)).is_err());
+        let foreground = budget.acquire(InferenceRole::Foreground, 1).unwrap();
+        drop(pending);
+        assert!(rx.recv_timeout(Duration::from_millis(50)).is_err());
+        drop(foreground);
+        rx.recv_timeout(Duration::from_secs(2)).unwrap();
+        worker.join().unwrap();
+    }
+    #[test]
+    fn panic_releases_threads_and_priority() {
+        let budget = InferenceBudget::new(1);
+        let shared = budget.clone();
+        assert!(std::panic::catch_unwind(move || {
+            let _pending = shared.pending_foreground();
+            let _permit = shared.acquire(InferenceRole::Foreground, 1).unwrap();
+            panic!("owned inference failure");
+        })
+        .is_err());
+        let _permit = budget.acquire(InferenceRole::Background, 1).unwrap();
+    }
+    #[test]
+    fn invalid_session_weight_fails_instead_of_oversubscribing() {
+        let budget = InferenceBudget::new(1);
+        assert!(budget.acquire(InferenceRole::Foreground, 2).is_err());
+        assert_eq!(budget.configure(0), 1);
+        assert!(InferenceBudget::new(usize::MAX).threads() <= 8);
+    }
+}

fixtures/mixed/sources/crates__searchi-service__src__routing.rs

@@ -0,0 +1,218 @@
+use super::*;
+
+/// Explicit semantic source channel, independent of filename/type filters.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SemanticOrigin {
+    All,
+    Text,
+    Image,
+    Video,
+    Audio,
+}
+impl SemanticOrigin {
+    fn source_type(self) -> Option<&'static str> {
+        match self {
+            Self::All => None,
+            Self::Text => Some("text"),
+            Self::Image => Some("image"),
+            Self::Video => Some("video"),
+            Self::Audio => Some("audio"),
+        }
+    }
+}
+
+/// A conservative intent hint, not a language understanding model. Explicit
+/// document/visual cues veto it; ordinary topic and instrument queries do not
+/// silently restrict modality. Word boundaries avoid filename substrings.
+fn preferred_origin(query: &str) -> Option<SemanticOrigin> {
+    if query.split_whitespace().count() == 1 && Path::new(query).extension().is_some() {
+        return None;
+    }
+    let lower = query.to_lowercase();
+    let words: Vec<_> = lower
+        .split(|c: char| !c.is_alphanumeric())
+        .filter(|s| !s.is_empty())
+        .collect();
+    let has = |values: &[&str]| words.iter().any(|word| values.contains(word));
+    if has(&[
+        "image",
+        "photo",
+        "picture",
+        "screenshot",
+        "diagram",
+        "document",
+        "pdf",
+        "text",
+        "paper",
+        "bild",
+        "foto",
+        "screenshot",
+        "dokument",
+        "textdatei",
+    ]) {
+        return None;
+    }
+    if has(&[
+        "audio",
+        "spoken",
+        "speech",
+        "recording",
+        "conversation",
+        "interview",
+        "podcast",
+        "sound",
+        "music",
+        "gesprochen",
+        "gespräch",
+        "gespraech",
+        "geräusch",
+        "geraeusch",
+        "musik",
+        "hörbar",
+        "hörbarer",
+        "hörbare",
+        "hoerbar",
+        "hoerbarer",
+        "hoerbare",
+        "tonaufnahme",
+    ]) {
+        return Some(SemanticOrigin::Audio);
+    }
+    if (has(&["someone", "speaker"])
+        && has(&[
+            "mentions",
+            "mentioned",
+            "says",
+            "said",
+            "speaks",
+            "speaking",
+        ]))
+        || (has(&["jemand", "sprecher", "sprecherin"])
+            && has(&["erwähnt", "erwaehnt", "sagt", "gesagt", "spricht"]))
+    {
+        return Some(SemanticOrigin::Audio);
+    }
+    None
+}
+
+impl HybridSearch {
+    pub(super) fn routed_semantic(
+        &mut self,
+        query: &str,
+        vector: &[f32],
+        limit: usize,
+        options: &SearchOptions,
+    ) -> Result<Vec<SearchResult>> {
+        let mut trace = TraceCollector::new(MetricsMode::Basic);
+        self.routed_semantic_traced(query, vector, limit, options, &mut trace, None)
+    }
+    pub(super) fn routed_semantic_traced(
+        &mut self,
+        query: &str,
+        vector: &[f32],
+        limit: usize,
+        options: &SearchOptions,
+        trace: &mut TraceCollector,
+        parent: Option<SpanToken>,
+    ) -> Result<Vec<SearchResult>> {
+        trace.timed("semantic.routing", parent, |trace, parent| {
+            if let Some(origin) = if options.code_only {
+                None
+            } else {
+                preferred_origin(query)
+            } {
+                trace.counter(parent, "audio_preference", 1);
+                let preferred = self.store.search_origin_traced(
+                    vector,
+                    &self.generation_fingerprint,
+                    limit,
+                    options,
+                    origin.source_type(),
+                    trace,
+                    parent,
+                )?;
+                if !preferred.is_empty() {
+                    return Ok(preferred);
+                }
+            }
+            trace.counter(parent, "general_lane", 1);
+            self.store.search_origin_traced(
+                vector,
+                &self.generation_fingerprint,
+                limit,
+                options,
+                None,
+                trace,
+                parent,
+            )
+        })
+    }
+
+    /// Query one actual source channel, or All to bypass automatic spoken/audio
+    /// preference. This never embeds files again or changes an active policy.
+    pub fn search_semantic_with_origin(
+        &mut self,
+        query: &str,
+        limit: usize,
+        options: &SearchOptions,
+        origin: SemanticOrigin,
+    ) -> Result<Vec<SearchResult>> {
+        let _ = self.lexical.search_with_options(query, 0, options)?;
+        if query.trim().is_empty() || limit == 0 {
+            return Ok(Vec::new());
+        }
+        let mut trace = TraceCollector::new(MetricsMode::Basic);
+        let vector = self.query_vector(query, options.code_only, &mut trace, None)?;
+        self.store.search_origin(
+            &vector,
+            &self.generation_fingerprint,
+            limit.min(1000),
+            options,
+            origin.source_type(),
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn persisted_custom_limits_require_identical_model_and_preprocessing() {
+        let expected = "pinned-model:pinned-media:preprocessing-v1";
+        assert!(reusable_policy(
+            expected,
+            &format!("{expected}:custom-240-2000-120-10000-2000-1800-14400000-17179869184")
+        ));
+        for active in ["other-model:pinned-media:preprocessing-v1:custom-240-2000-120-10000-2000-1800-14400000-17179869184","pinned-model:pinned-media:preprocessing-v2:custom-240-2000-120-10000-2000-1800-14400000-17179869184","pinned-model:pinned-media:preprocessing-v1:custom-unknown"] {assert!(!reusable_policy(expected,active));}
+    }
+    #[test]
+    fn explicit_spoken_intent_is_conservative_and_bilingual() {
+        for query in [
+            "where someone mentions a contract",
+            "a speaker says the date",
+            "spoken discussion of ecology",
+            "audio of a waterfall",
+            "jemand erwähnt das Budget",
+            "Wo spricht jemand über Kündigung?",
+            "Gespräch über die neue Wohnung",
+            "hörbarer Regen",
+        ] {
+            assert_eq!(
+                preferred_origin(query),
+                Some(SemanticOrigin::Audio),
+                "{query}"
+            );
+        }
+        for query in [
+            "contract deadline",
+            "violin solo",
+            "photo of someone speaking",
+            "Bild von einem Sprecher",
+            "a document that someone mentioned",
+            "speech.png",
+            "a screenshot of an audio waveform",
+        ] {
+            assert_eq!(preferred_origin(query), None, "{query}");
+        }
+    }
+}

fixtures/mixed/sources/crates__searchi-service__src__store__incremental.rs

@@ -0,0 +1,967 @@
+//! Completed-source manifests and resumable staging. Legacy generations stay
+//! searchable but cannot supply vectors without explicit completion evidence.
+use super::*;
+
+#[derive(Debug, Clone)]
+pub(crate) struct SourceTask {
+    pub key: String,
+    pub path: String,
+    /// `text` for one extracted passage, `media` for the entire media file.
+    pub source_type: String,
+    pub source_hash: String,
+    pub title_hash: String,
+    pub policy: String,
+    pub passage_id: i64,
+}
+
+pub(super) fn migrate(connection: &Connection) -> Result<()> {
+    for (table, name, declaration) in [
+        (
+            "semantic_generations",
+            "incremental",
+            "INTEGER NOT NULL DEFAULT 0",
+        ),
+        ("semantic_chunks", "task_key", "TEXT"),
+        (
+            "semantic_chunks",
+            "file_change_revision",
+            "INTEGER NOT NULL DEFAULT 0",
+        ),
+    ] {
+        let exists: bool = connection.query_row(
+            "SELECT EXISTS(SELECT 1 FROM pragma_table_info(?1) WHERE name=?2)",
+            params![table, name],
+            |r| r.get(0),
+        )?;
+        if !exists {
+            connection.execute_batch(&format!(
+                "ALTER TABLE {table} ADD COLUMN {name} {declaration}"
+            ))?;
+        }
+    }
+    connection.execute_batch("CREATE TABLE IF NOT EXISTS semantic_tasks(
+        generation INTEGER NOT NULL REFERENCES semantic_generations(id) ON DELETE CASCADE,
+        task_key TEXT NOT NULL,path TEXT NOT NULL,source_type TEXT NOT NULL,
+        source_hash TEXT NOT NULL,title_hash TEXT NOT NULL,policy TEXT NOT NULL,
+        passage_id INTEGER NOT NULL,state TEXT NOT NULL CHECK(state IN ('pending','running','done','failed')),
+        attempts INTEGER NOT NULL DEFAULT 0,expected_chunks INTEGER,message TEXT,
+        PRIMARY KEY(generation,task_key));
+        CREATE INDEX IF NOT EXISTS semantic_task_reuse ON semantic_tasks(source_hash,title_hash,policy,source_type,state);
+        CREATE INDEX IF NOT EXISTS semantic_chunk_task ON semantic_chunks(generation,task_key);
+        UPDATE semantic_meta SET schema_version=8 WHERE singleton=1;")?;
+    Ok(())
+}
+
+pub(super) fn validate_completion(connection: &Connection, generation: i64) -> Result<()> {
+    let incomplete: bool = connection.query_row("SELECT EXISTS(SELECT 1 FROM semantic_tasks t WHERE generation=?1 AND
+        (state NOT IN ('done','failed') OR (state='done' AND (expected_chunks IS NULL OR expected_chunks!=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key))) OR
+        (state='failed' AND EXISTS(SELECT 1 FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key))))",[generation], |r|r.get(0))?;
+    ensure!(
+        !incomplete,
+        "Semantic generation contains incomplete source tasks"
+    );
+    let unowned: bool = connection.query_row("SELECT EXISTS(SELECT 1 FROM semantic_chunks c LEFT JOIN semantic_tasks t ON t.generation=c.generation AND t.task_key=c.task_key WHERE c.generation=?1 AND (t.state IS NULL OR t.state!='done'))",[generation],|r|r.get(0))?;
+    ensure!(
+        !unowned,
+        "Semantic generation contains chunks without completed sources"
+    );
+    Ok(())
+}
+
+impl Store {
+    pub(crate) fn revision(&self) -> Result<i64> {
+        Ok(self.connection.query_row(
+            "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
+            [],
+            |r| r.get(0),
+        )?)
+    }
+
+    /// Caller must retain the OS lock through reconciliation and publication.
+    pub(crate) fn start_or_resume(
+        &mut self,
+        fingerprint: &str,
+        lock: &BuildLock,
+    ) -> Result<(i64, bool)> {
+        ensure!(
+            lock.1 == self.lock_path,
+            "Build lock belongs to a different index"
+        );
+        ensure!(
+            !fingerprint.is_empty() && fingerprint.len() <= 4096,
+            "Invalid generation fingerprint"
+        );
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let generation: Option<i64>=tx.query_row("SELECT id FROM semantic_generations WHERE id!=COALESCE((SELECT active_generation FROM semantic_meta WHERE singleton=1),-1) AND fingerprint=?1 AND dimensions=?2 AND incremental=1 ORDER BY id DESC LIMIT 1",params![fingerprint,self.dimensions as i64],|r|r.get(0)).optional()?;
+        let resumed = generation.is_some();
+        let id = if let Some(id) = generation {
+            id
+        } else {
+            tx.execute("INSERT INTO semantic_generations(fingerprint,dimensions,corpus_revision,incremental) SELECT ?1,?2,corpus_revision,1 FROM semantic_meta WHERE singleton=1",params![fingerprint,self.dimensions as i64])?;
+            tx.last_insert_rowid()
+        };
+        tx.execute("DELETE FROM semantic_generations WHERE id!=?1 AND id!=COALESCE((SELECT active_generation FROM semantic_meta WHERE singleton=1),-1)",[id])?;
+        // Untagged chunks are writes interrupted before the source transaction.
+        tx.execute("UPDATE semantic_tasks SET state='pending',expected_chunks=NULL WHERE generation=?1 AND state='done' AND
+            (expected_chunks IS NULL OR expected_chunks!=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=semantic_tasks.generation AND c.task_key=semantic_tasks.task_key))",[id])?;
+        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND NOT EXISTS(SELECT 1 FROM semantic_tasks t WHERE t.generation=semantic_chunks.generation AND t.task_key=semantic_chunks.task_key AND t.state='done')",[id])?;
+        tx.execute("UPDATE semantic_tasks SET state='pending',expected_chunks=NULL,message=NULL WHERE generation=?1 AND state!='done'",[id])?;
+        tx.commit()?;
+        Ok((id, resumed))
+    }
+
+    pub(crate) fn stage_task(&mut self, generation: i64, task: &SourceTask) -> Result<bool> {
+        validate_task(task)?;
+        let tx = self.connection.transaction()?;
+        let same:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM semantic_tasks WHERE generation=?1 AND task_key=?2 AND source_hash=?3 AND title_hash=?4 AND policy=?5 AND source_type=?6 AND state='done')",params![generation,task.key,task.source_hash,task.title_hash,task.policy,task.source_type],|r|r.get(0))?;
+        if !same {
+            tx.execute(
+                "DELETE FROM semantic_chunks WHERE generation=?1 AND task_key=?2",
+                params![generation, task.key],
+            )?;
+            tx.execute("INSERT INTO semantic_tasks(generation,task_key,path,source_type,source_hash,title_hash,policy,passage_id,state,attempts)
+                VALUES(?1,?2,?3,?4,?5,?6,?7,?8,'running',1)
+                ON CONFLICT(generation,task_key) DO UPDATE SET path=excluded.path,source_type=excluded.source_type,source_hash=excluded.source_hash,title_hash=excluded.title_hash,policy=excluded.policy,passage_id=excluded.passage_id,state='running',attempts=semantic_tasks.attempts+1,expected_chunks=NULL,message=NULL",params![generation,task.key,task.path,task.source_type,task.source_hash,task.title_hash,task.policy,task.passage_id])?;
+        }
+        tx.commit()?;
+        Ok(same)
+    }
+
+    pub(crate) fn complete_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        expected_chunks: usize,
+    ) -> Result<()> {
+        ensure!(
+            !task.source_hash.is_empty(),
+            "Cannot complete a source without verified identity"
+        );
+        ensure!(
+            expected_chunks <= MAX_SCAN_CHUNKS,
+            "Source exceeds semantic chunk budget"
+        );
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let count:i64=tx.query_row("SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND path=?2 AND ((?3='text' AND source_type='text' AND passage_id=?4) OR (?3='media' AND source_type!='text'))",params![generation,task.path,task.source_type,task.passage_id],|r|r.get(0))?;
+        ensure!(
+            usize::try_from(count)? == expected_chunks,
+            "Incomplete source chunk membership"
+        );
+        let total: usize = tx.query_row(
+            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+            [generation],
+            |r| r.get(0),
+        )?;
+        ensure!(
+            total <= MAX_SCAN_CHUNKS,
+            "Generation exceeds semantic chunk budget"
+        );
+        tx.execute("UPDATE semantic_chunks SET task_key=?5 WHERE generation=?1 AND path=?2 AND ((?3='text' AND source_type='text' AND passage_id=?4) OR (?3='media' AND source_type!='text'))",params![generation,task.path,task.source_type,task.passage_id,task.key])?;
+        let changed=tx.execute("UPDATE semantic_tasks SET state='done',expected_chunks=?3,path=?4,passage_id=?5,message=NULL WHERE generation=?1 AND task_key=?2 AND source_hash=?6 AND title_hash=?7 AND policy=?8 AND source_type=?9",params![generation,task.key,count,task.path,task.passage_id,task.source_hash,task.title_hash,task.policy,task.source_type])?;
+        ensure!(changed == 1, "Missing matching source task");
+        if task.source_type == "media" {
+            tx.execute(
+                "DELETE FROM semantic_media_errors WHERE generation=?1 AND path=?2",
+                params![generation, task.path],
+            )?;
+        }
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn fail_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        message: &str,
+    ) -> Result<()> {
+        let tx = self.connection.transaction()?;
+        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND (task_key=?2 OR (path=?3 AND ((?4='text' AND source_type='text' AND passage_id=?5) OR (?4='media' AND source_type!='text'))))",params![generation,task.key,task.path,task.source_type,task.passage_id])?;
+        tx.execute("UPDATE semantic_tasks SET state='failed',expected_chunks=NULL,message=?3 WHERE generation=?1 AND task_key=?2",params![generation,task.key,message.chars().take(2048).collect::<String>()])?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    /// Reconcile completed staging groups before any encoder call. A bounded
+    /// SQL snapshot preserves renamed donors (including swaps and duplicates)
+    /// while rebuilding only current compatible memberships. Active vectors
+    /// are never touched. Full range/vector validation still happens on reuse.
+    pub(crate) fn prune_before_inference(
+        &mut self,
+        generation: i64,
+        tasks: &[SourceTask],
+        revision: i64,
+    ) -> Result<()> {
+        ensure!(
+            tasks.len() <= MAX_SCAN_CHUNKS,
+            "Generation exceeds source task budget"
+        );
+        for task in tasks {
+            validate_task(task)?;
+        }
+        // The bounded snapshot can still contain 100K x 768d vectors. Keep it
+        // SQLite-backed on disk rather than turning that into a Rust/heap copy.
+        self.connection.pragma_update(None, "temp_store", "FILE")?;
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let current: i64 = tx.query_row(
+            "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
+            [],
+            |r| r.get(0),
+        )?;
+        ensure!(
+            current == revision,
+            "Indexed corpus changed during semantic discovery; retry required"
+        );
+        let active: bool = tx.query_row(
+            "SELECT active_generation=?1 FROM semantic_meta WHERE singleton=1",
+            [generation],
+            |r| r.get::<_, Option<bool>>(0).map(|v| v.unwrap_or(false)),
+        )?;
+        ensure!(!active, "Cannot prune the active semantic generation");
+        let existing: usize = tx.query_row(
+            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+            [generation],
+            |r| r.get(0),
+        )?;
+        ensure!(
+            existing <= MAX_SCAN_CHUNKS,
+            "Staging generation exceeds semantic chunk budget"
+        );
+        tx.execute_batch("CREATE TEMP TABLE IF NOT EXISTS semantic_discovery_tasks(task_key TEXT PRIMARY KEY,path TEXT NOT NULL,source_type TEXT NOT NULL,source_hash TEXT NOT NULL,title_hash TEXT NOT NULL,policy TEXT NOT NULL,passage_id INTEGER NOT NULL);
+            DELETE FROM semantic_discovery_tasks;
+            CREATE TEMP TABLE IF NOT EXISTS semantic_prune_tasks AS SELECT * FROM semantic_tasks WHERE 0;
+            CREATE TEMP TABLE IF NOT EXISTS semantic_prune_chunks AS SELECT * FROM semantic_chunks WHERE 0;
+            DELETE FROM semantic_prune_tasks; DELETE FROM semantic_prune_chunks;")?;
+        for task in tasks {
+            tx.execute(
+                "INSERT INTO semantic_discovery_tasks VALUES(?1,?2,?3,?4,?5,?6,?7)",
+                params![
+                    task.key,
+                    task.path,
+                    task.source_type,
+                    task.source_hash,
+                    task.title_hash,
+                    task.policy,
+                    task.passage_id
+                ],
+            )?;
+        }
+        tx.execute("INSERT INTO semantic_prune_tasks SELECT * FROM semantic_tasks t WHERE generation=?1 AND state='done' AND source_hash!='' AND expected_chunks IS NOT NULL AND expected_chunks=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key)",[generation])?;
+        tx.execute(
+            "INSERT INTO semantic_prune_chunks SELECT * FROM semantic_chunks WHERE generation=?1",
+            [generation],
+        )?;
+        tx.execute_batch("CREATE UNIQUE INDEX IF NOT EXISTS temp.semantic_prune_task_key ON semantic_prune_tasks(task_key);
+            CREATE INDEX IF NOT EXISTS temp.semantic_prune_reuse ON semantic_prune_tasks(source_hash,title_hash,policy,source_type);
+            CREATE INDEX IF NOT EXISTS temp.semantic_prune_membership ON semantic_prune_chunks(task_key);")?;
+        let mut mappings = Vec::new();
+        {
+            let mut statement = tx.prepare("SELECT d.task_key,COALESCE((SELECT old.task_key FROM semantic_prune_tasks old WHERE old.task_key=d.task_key AND old.source_hash=d.source_hash AND old.title_hash=d.title_hash AND old.policy=d.policy AND old.source_type=d.source_type),(SELECT old.task_key FROM semantic_prune_tasks old WHERE old.source_hash=d.source_hash AND old.title_hash=d.title_hash AND old.policy=d.policy AND old.source_type=d.source_type ORDER BY old.task_key LIMIT 1)) FROM semantic_discovery_tasks d")?;
+            let rows = statement.query_map([], |r| {
+                Ok((r.get::<_, String>(0)?, r.get::<_, Option<String>>(1)?))
+            })?;
+            for row in rows {
+                let (key, donor) = row?;
+                if let Some(donor) = donor {
+                    mappings.push((key, donor));
+                }
+            }
+        }
+        let mut total = 0usize;
+        for (_, donor) in &mappings {
+            let count: usize = tx.query_row(
+                "SELECT expected_chunks FROM semantic_prune_tasks WHERE task_key=?1",
+                [donor],
+                |r| r.get(0),
+            )?;
+            total = total
+                .checked_add(count)
+                .context("Semantic chunk count overflow")?;
+            ensure!(
+                total <= MAX_SCAN_CHUNKS,
+                "Current completed sources exceed semantic chunk budget"
+            );
+        }
+        tx.execute(
+            "DELETE FROM semantic_chunks WHERE generation=?1",
+            [generation],
+        )?;
+        tx.execute(
+            "DELETE FROM semantic_tasks WHERE generation=?1",
+            [generation],
+        )?;
+        for (key, donor) in mappings {
+            tx.execute("INSERT INTO semantic_tasks(generation,task_key,path,source_type,source_hash,title_hash,policy,passage_id,state,attempts,expected_chunks,message)
+                SELECT ?1,d.task_key,d.path,d.source_type,d.source_hash,d.title_hash,d.policy,d.passage_id,'done',old.attempts,old.expected_chunks,NULL FROM semantic_discovery_tasks d JOIN semantic_prune_tasks old ON old.task_key=?3 WHERE d.task_key=?2",params![generation,key,donor])?;
+            tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key,file_change_revision)
+                SELECT ?1,d.passage_id,c.ordinal,d.path,c.kind,c.number,c.content_hash,c.start_byte,c.end_byte,c.snippet,c.vector,c.source_type,c.file_size,c.file_mtime,d.task_key,COALESCE((SELECT change_revision FROM files WHERE path=d.path),c.file_change_revision) FROM semantic_prune_chunks c JOIN semantic_discovery_tasks d ON d.task_key=?2 WHERE c.task_key=?3",params![generation,key,donor])?;
+        }
+        tx.execute(
+            "DELETE FROM semantic_media_errors WHERE generation=?1",
+            [generation],
+        )?;
+        tx.execute_batch("DELETE FROM semantic_prune_tasks; DELETE FROM semantic_prune_chunks; DELETE FROM semantic_discovery_tasks;")?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn reconcile_tasks(
+        &mut self,
+        generation: i64,
+        keys: &[String],
+        revision: i64,
+    ) -> Result<()> {
+        ensure!(
+            keys.len() <= MAX_SCAN_CHUNKS,
+            "Generation exceeds source task budget"
+        );
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let current: i64 = tx.query_row(
+            "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
+            [],
+            |r| r.get(0),
+        )?;
+        ensure!(
+            current == revision,
+            "Indexed corpus changed during semantic reconciliation; retry required"
+        );
+        tx.execute_batch("CREATE TEMP TABLE IF NOT EXISTS semantic_current_tasks(task_key TEXT PRIMARY KEY); DELETE FROM semantic_current_tasks;")?;
+        for key in keys {
+            tx.execute(
+                "INSERT OR IGNORE INTO semantic_current_tasks VALUES(?1)",
+                [key],
+            )?;
+        }
+        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND (task_key IS NULL OR task_key NOT IN (SELECT task_key FROM semantic_current_tasks))",[generation])?;
+        tx.execute("DELETE FROM semantic_tasks WHERE generation=?1 AND task_key NOT IN (SELECT task_key FROM semantic_current_tasks)",[generation])?;
+        tx.execute("DELETE FROM semantic_media_errors WHERE generation=?1 AND path NOT IN (SELECT path FROM semantic_tasks WHERE generation=?1)",[generation])?;
+        tx.execute(
+            "UPDATE semantic_generations SET corpus_revision=?2 WHERE id=?1",
+            params![generation, revision],
+        )?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    #[cfg(test)]
+    pub(crate) fn activate_checked(&mut self, generation: i64, revision: i64) -> Result<()> {
+        self.activate_checked_with_validation(generation, revision, || Ok(()))
+    }
+
+    /// Validate live sources after graph preparation, outside the SQLite write
+    /// transaction. Rejection preserves both the active reader and staging.
+    pub(crate) fn activate_checked_with_validation(
+        &mut self,
+        generation: i64,
+        revision: i64,
+        validate_sources: impl FnOnce() -> Result<()>,
+    ) -> Result<()> {
+        validate_completion(&self.connection, generation)?;
+        self.activate_revision(generation, Some(revision), validate_sources)
+    }
+}
+
+fn validate_task(task: &SourceTask) -> Result<()> {
+    ensure!(
+        !task.key.is_empty()
+            && task.key.len() <= 256
+            && !task.path.is_empty()
+            && task.path.len() <= 32768,
+        "Invalid source task identity"
+    );
+    ensure!(
+        matches!(task.source_type.as_str(), "text" | "media"),
+        "Invalid source task type"
+    );
+    ensure!(
+        (!task.source_hash.is_empty() || task.source_type == "media")
+            && task.source_hash.len() <= 256
+            && task.title_hash.len() <= 256
+            && !task.policy.is_empty()
+            && task.policy.len() <= 4096,
+        "Invalid source completion policy"
+    );
+    Ok(())
+}
+
+impl Store {
+    fn reusable_task(
+        &self,
+        generation: i64,
+        task: &SourceTask,
+    ) -> Result<Option<(i64, String, usize)>> {
+        Ok(self.connection.query_row("SELECT t.generation,t.task_key,t.expected_chunks FROM semantic_tasks t JOIN semantic_generations g ON g.id=t.generation JOIN semantic_generations destination ON destination.id=?1
+            WHERE t.source_hash=?2 AND t.title_hash=?3 AND t.policy=?4 AND t.source_type=?5 AND t.state='done' AND t.expected_chunks IS NOT NULL
+            AND g.fingerprint=destination.fingerprint AND g.dimensions=destination.dimensions
+            AND t.expected_chunks=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key)
+            ORDER BY (t.generation=?1) DESC,t.generation DESC LIMIT 1",params![generation,task.source_hash,task.title_hash,task.policy,task.source_type],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?)
+    }
+
+    /// No tokenizer or neural call: only a completed, policy-compatible group
+    /// whose persisted ranges still equal the current extracted bytes qualifies.
+    pub(crate) fn reuse_text_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        passage: &IndexedPassage,
+    ) -> Result<Option<usize>> {
+        ensure!(
+            task.source_type == "text"
+                && task.passage_id == passage.id
+                && task.path == passage.path,
+            "Text task mapping differs from current source"
+        );
+        ensure!(
+            task.source_hash == content_hash(&passage.text),
+            "Text task hash differs from source bytes"
+        );
+        let Some((source, key, count)) = self.reusable_task(generation, task)? else {
+            return Ok(None);
+        };
+        ensure!(
+            count <= MAX_SCAN_CHUNKS,
+            "Reusable source exceeds chunk budget"
+        );
+        {
+            let mut stmt=self.connection.prepare("SELECT start_byte,end_byte,snippet,vector,source_type,ordinal,content_hash FROM semantic_chunks WHERE generation=?1 AND task_key=?2 ORDER BY ordinal")?;
+            let mut rows = stmt.query(params![source, key])?;
+            let mut seen = 0usize;
+            while let Some(row) = rows.next()? {
+                let start: usize = row.get(0)?;
+                let end: usize = row.get(1)?;
+                let snippet: String = row.get(2)?;
+                ensure!(
+                    passage.text.get(start..end) == Some(snippet.as_str()),
+                    "Reusable text range differs from current source"
+                );
+                ensure!(
+                    row.get::<_, String>(4)? == "text"
+                        && row.get::<_, usize>(5)? == seen
+                        && row.get::<_, String>(6)? == task.source_hash,
+                    "Corrupt completed text source membership"
+                );
+                validate_vector(&row.get::<_, Vec<u8>>(3)?, self.dimensions)?;
+                seen += 1;
+            }
+            ensure!(seen == count, "Reusable text completion count differs");
+        }
+        self.copy_completed(
+            generation,
+            task,
+            source,
+            &key,
+            count,
+            Some(&passage.source),
+            None,
+        )?;
+        Ok(Some(count))
+    }
+
+    pub(crate) fn reuse_media_source(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        file_id: i64,
+        identity: &searchi_media::MediaIdentity,
+    ) -> Result<Option<usize>> {
+        ensure!(
+            task.source_type == "media"
+                && task.passage_id == -file_id
+                && task.path == identity.path.to_string_lossy()
+                && task.source_hash == identity.sha256,
+            "Media task mapping differs from current source"
+        );
+        let Some((source, key, count)) = self.reusable_task(generation, task)? else {
+            return Ok(None);
+        };
+        ensure!(
+            count <= MAX_SCAN_CHUNKS,
+            "Reusable source exceeds chunk budget"
+        );
+        {
+            let mut stmt=self.connection.prepare("SELECT vector,source_type,ordinal,content_hash FROM semantic_chunks WHERE generation=?1 AND task_key=?2 ORDER BY ordinal")?;
+            let mut rows = stmt.query(params![source, key])?;
+            let mut seen = 0usize;
+            while let Some(row) = rows.next()? {
+                ensure!(
+                    matches!(
+                        row.get::<_, String>(1)?.as_str(),
+                        "image" | "audio" | "video"
+                    ) && row.get::<_, usize>(2)? == seen
+                        && row.get::<_, String>(3)? == task.source_hash,
+                    "Corrupt completed media source membership"
+                );
+                validate_vector(&row.get::<_, Vec<u8>>(0)?, self.dimensions)?;
+                seen += 1;
+            }
+            ensure!(seen == count, "Reusable media completion count differs");
+        }
+        self.copy_completed(generation, task, source, &key, count, None, Some(identity))?;
+        Ok(Some(count))
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn copy_completed(
+        &mut self,
+        generation: i64,
+        task: &SourceTask,
+        source: i64,
+        key: &str,
+        count: usize,
+        location: Option<&SourceLocation>,
+        identity: Option<&searchi_media::MediaIdentity>,
+    ) -> Result<()> {
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let total:usize=tx.query_row("SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND (task_key IS NULL OR task_key!=?2)",params![generation,task.key],|r|r.get(0))?;
+        ensure!(
+            total.saturating_add(count) <= MAX_SCAN_CHUNKS,
+            "Generation exceeds semantic chunk budget"
+        );
+        // A temporary SQL table allows remapping a completed staging task onto
+        // itself without retaining its old citations or allocating all vectors.
+        tx.execute_batch("CREATE TEMP TABLE IF NOT EXISTS semantic_reuse_chunks AS SELECT * FROM semantic_chunks WHERE 0; DELETE FROM semantic_reuse_chunks;")?;
+        tx.execute("INSERT INTO semantic_reuse_chunks SELECT * FROM semantic_chunks WHERE generation=?1 AND task_key=?2",params![source,key])?;
+        tx.execute(
+            "DELETE FROM semantic_chunks WHERE generation=?1 AND task_key=?2",
+            params![generation, task.key],
+        )?;
+        tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key,file_change_revision)
+            SELECT ?1,?2,ordinal,?3,COALESCE(?4,kind),COALESCE(?5,number),content_hash,start_byte,end_byte,snippet,vector,source_type,COALESCE(?6,file_size),COALESCE(?7,file_mtime),?8,COALESCE((SELECT change_revision FROM files WHERE path=?3),file_change_revision) FROM semantic_reuse_chunks",params![generation,task.passage_id,task.path,location.map(|s|s.kind.as_str()),location.map(|s|s.number),identity.map(|i|i.bytes as i64),identity.map(|i|i.modified_ns),task.key])?;
+        let changed=tx.execute("UPDATE semantic_tasks SET state='done',expected_chunks=?3,path=?4,passage_id=?5,message=NULL WHERE generation=?1 AND task_key=?2 AND source_hash=?6 AND title_hash=?7 AND policy=?8 AND source_type=?9",params![generation,task.key,count as i64,task.path,task.passage_id,task.source_hash,task.title_hash,task.policy,task.source_type])?;
+        ensure!(changed == 1, "Missing source task during reuse");
+        if task.source_type == "media" {
+            tx.execute(
+                "DELETE FROM semantic_media_errors WHERE generation=?1 AND path=?2",
+                params![generation, task.path],
+            )?;
+        }
+        tx.execute_batch("DELETE FROM semantic_reuse_chunks;")?;
+        tx.commit()?;
+        Ok(())
+    }
+}
+
+fn validate_vector(bytes: &[u8], dimensions: usize) -> Result<()> {
+    ensure!(
+        bytes.len() == dimensions * 4,
+        "Corrupt reusable vector length"
+    );
+    let mut norm = 0.0f32;
+    for value in bytes.as_chunks::<4>().0 {
+        let value = f32::from_le_bytes(*value);
+        ensure!(value.is_finite(), "Corrupt reusable vector value");
+        norm += value * value;
+    }
+    ensure!(
+        norm.is_finite() && (norm.sqrt() - 1.0).abs() < 1e-3,
+        "Reusable vector is not normalized"
+    );
+    Ok(())
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use searchi_core::SearchIndex;
+
+    // Storage-only vectors exercise durability and membership. They are not
+    // simulated production inference or evidence of semantic relevance.
+    fn vector() -> Vec<f32> {
+        vec![1.0, 0.0]
+    }
+    fn task(p: &IndexedPassage) -> SourceTask {
+        SourceTask {
+            key: format!("text-{}-{}", p.path, p.source.number),
+            path: p.path.clone(),
+            source_type: "text".into(),
+            source_hash: content_hash(&p.text),
+            title_hash: content_hash(&p.filename),
+            policy: "storage-test-window-policy".into(),
+            passage_id: p.id,
+        }
+    }
+    fn insert_complete(
+        store: &mut Store,
+        g: i64,
+        p: &IndexedPassage,
+        t: &SourceTask,
+    ) -> Result<()> {
+        store.stage_task(g, t)?;
+        store.insert(
+            g,
+            p,
+            0,
+            &[TextWindow {
+                text: p.text.clone(),
+                start_byte: 0,
+                end_byte: p.text.len(),
+            }],
+            &[vector()],
+        )?;
+        store.complete_source(g, t, 1)
+    }
+
+    #[test]
+    fn pre_inference_pruning_preserves_renamed_donors_at_the_chunk_ceiling() -> Result<()> {
+        // Actual 100K membership boundary; vectors are storage-only fixtures.
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("notes.txt"), "complete source")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let p = lexical.passage_batch(0, 1)?.remove(0);
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        let active = store.start_or_resume("prune-policy", &lock)?.0;
+        insert_complete(&mut store, active, &p, &task(&p))?;
+        store.activate_checked(active, store.revision()?)?;
+        let staging = store.start_or_resume("prune-policy", &lock)?.0;
+        let original = task(&p);
+        insert_complete(&mut store, staging, &p, &original)?;
+        store.connection.execute("WITH RECURSIVE ord(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM ord WHERE n+1<?2)
+            INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key)
+            SELECT generation,passage_id,n,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key FROM semantic_chunks CROSS JOIN ord WHERE generation=?1 AND ordinal=0",params![staging,MAX_SCAN_CHUNKS as i64])?;
+        store.complete_source(staging, &original, MAX_SCAN_CHUNKS)?;
+        let moved_dir = root.join("moved");
+        std::fs::create_dir(&moved_dir)?;
+        std::fs::rename(root.join("notes.txt"), moved_dir.join("notes.txt"))?;
+        lexical.index_folder(&root)?;
+        let moved = lexical.passage_batch(0, 1)?.remove(0);
+        let moved_task = task(&moved);
+        let revision = store.revision()?;
+        assert!(store
+            .prune_before_inference(staging, std::slice::from_ref(&moved_task), revision - 1)
+            .is_err());
+        let old_count: usize = store.connection.query_row(
+            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+            [staging],
+            |r| r.get(0),
+        )?;
+        assert_eq!(
+            old_count, MAX_SCAN_CHUNKS,
+            "Revision failure must preserve staging"
+        );
+        store.prune_before_inference(staging, std::slice::from_ref(&moved_task), revision)?;
+        store.stage_task(staging, &moved_task)?;
+        assert_eq!(
+            store.reuse_text_source(staging, &moved_task, &moved)?,
+            Some(MAX_SCAN_CHUNKS)
+        );
+        let (count, old_paths): (usize, usize) = store.connection.query_row(
+            "SELECT count(*),sum(path!=?2) FROM semantic_chunks WHERE generation=?1",
+            params![staging, moved.path],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )?;
+        assert_eq!((count, old_paths), (MAX_SCAN_CHUNKS, 0));
+        let mut duplicate = moved_task.clone();
+        duplicate.key.push_str("-another-path");
+        duplicate.path.push_str("-another-path");
+        assert!(
+            store
+                .prune_before_inference(staging, &[moved_task.clone(), duplicate], revision)
+                .is_err(),
+            "True current membership over budget must reject atomically"
+        );
+        let count: usize = store.connection.query_row(
+            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+            [staging],
+            |r| r.get(0),
+        )?;
+        assert_eq!(count, MAX_SCAN_CHUNKS);
+        std::fs::write(
+            moved_dir.join("notes.txt"),
+            "replacement source with new content",
+        )?;
+        lexical.index_folder(&root)?;
+        let replacement = lexical.passage_batch(0, 1)?.remove(0);
+        let replacement_task = task(&replacement);
+        store.prune_before_inference(
+            staging,
+            std::slice::from_ref(&replacement_task),
+            store.revision()?,
+        )?;
+        let count: usize = store.connection.query_row(
+            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+            [staging],
+            |r| r.get(0),
+        )?;
+        assert_eq!(
+            count, 0,
+            "Obsolete ceiling-sized group must be removed before replacement inference"
+        );
+        insert_complete(&mut store, staging, &replacement, &replacement_task)?;
+        assert_eq!(store.active_fingerprint()?.as_deref(), Some("prune-policy"));
+        let active_count: usize = store.connection.query_row(
+            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+            [active],
+            |r| r.get(0),
+        )?;
+        assert_eq!(active_count, 1, "Pruning must never remove active vectors");
+        Ok(())
+    }
+
+    #[test]
+    fn resume_retains_completed_groups_and_discards_unfinished_vectors() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("first.txt"), "complete source")?;
+        std::fs::write(root.join("second.txt"), "interrupted source")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let passages = lexical.passage_batch(0, 64)?;
+        let first = task(&passages[0]);
+        let second = task(&passages[1]);
+        let generation;
+        {
+            let mut store = Store::open(&db, 2)?;
+            let lock = store.build_lock()?;
+            generation = store.start_or_resume("storage-only-model", &lock)?.0;
+            insert_complete(&mut store, generation, &passages[0], &first)?;
+            store.stage_task(generation, &second)?;
+            store.insert(
+                generation,
+                &passages[1],
+                0,
+                &[TextWindow {
+                    text: passages[1].text.clone(),
+                    start_byte: 0,
+                    end_byte: passages[1].text.len(),
+                }],
+                &[vector()],
+            )?;
+            assert!(store
+                .activate_checked(generation, store.revision()?)
+                .is_err());
+        }
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        assert_eq!(
+            store.start_or_resume("storage-only-model", &lock)?,
+            (generation, true)
+        );
+        assert_eq!(
+            store.connection.query_row(
+                "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+                [generation],
+                |r| r.get::<_, usize>(0)
+            )?,
+            1
+        );
+        store.stage_task(generation, &first)?;
+        assert_eq!(
+            store.reuse_text_source(generation, &first, &passages[0])?,
+            Some(1)
+        );
+        assert!(store
+            .reuse_text_source(generation, &second, &passages[1])?
+            .is_none());
+        assert!(store
+            .activate_checked(generation, store.revision()?)
+            .is_err());
+        store.reconcile_tasks(generation, &[first.key], store.revision()?)?;
+        store.activate_checked(generation, store.revision()?)?;
+        Ok(())
+    }
+
+    #[test]
+    fn legacy_vectors_remain_searchable_but_do_not_qualify_for_reuse() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("notes.txt"), "legacy searchable source")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let p = lexical.passage_batch(0, 1)?.remove(0);
+        let mut store = Store::open(&db, 2)?;
+        let legacy = store.begin("legacy-policy")?;
+        store.insert(
+            legacy,
+            &p,
+            0,
+            &[TextWindow {
+                text: p.text.clone(),
+                start_byte: 0,
+                end_byte: p.text.len(),
+            }],
+            &[vector()],
+        )?;
+        store.activate(legacy)?;
+        // Reopen a schema6 DB with legacy active chunks and no manifests.
+        store.connection.execute_batch(
+            "DROP INDEX semantic_chunk_task; ALTER TABLE semantic_chunks DROP COLUMN task_key;
+            DROP TABLE semantic_tasks; ALTER TABLE semantic_generations DROP COLUMN incremental;
+            UPDATE semantic_meta SET schema_version=6;",
+        )?;
+        drop(store);
+        let mut store = Store::open(&db, 2)?;
+        assert_eq!(
+            store
+                .search(&vector(), "legacy-policy", 1, &SearchOptions::default())?
+                .len(),
+            1
+        );
+        let lock = store.build_lock()?;
+        let (g, _) = store.start_or_resume("legacy-policy", &lock)?;
+        let t = task(&p);
+        store.stage_task(g, &t)?;
+        assert_eq!(store.reuse_text_source(g, &t, &p)?, None);
+        assert_eq!(
+            store.active_fingerprint()?.as_deref(),
+            Some("legacy-policy")
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn late_source_validation_rejects_publication_without_holding_write_lock() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("notes.txt"), "source identity before graph build")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let passage = lexical.passage_batch(0, 1)?.remove(0);
+        let source = task(&passage);
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        let (active, _) = store.start_or_resume("old-policy", &lock)?;
+        insert_complete(&mut store, active, &passage, &source)?;
+        store.activate_checked(active, store.revision()?)?;
+        let (staging, _) = store.start_or_resume("new-policy", &lock)?;
+        insert_complete(&mut store, staging, &passage, &source)?;
+        let revision = store.revision()?;
+        let mut called = false;
+        let error = store
+            .activate_checked_with_validation(staging, revision, || {
+                called = true;
+                // An independent write succeeds only when validation runs outside
+                // the publishing connection's Immediate transaction.
+                let other = rusqlite::Connection::open(&db)?;
+                other.busy_timeout(std::time::Duration::ZERO)?;
+                other.execute_batch("BEGIN IMMEDIATE; COMMIT;")?;
+                std::fs::write(root.join("notes.txt"), "direct filesystem mutation")?;
+                anyhow::bail!("source changed after graph preparation")
+            })
+            .unwrap_err();
+        assert!(called);
+        assert!(error
+            .to_string()
+            .contains("source changed after graph preparation"));
+        assert_eq!(store.active_fingerprint()?.as_deref(), Some("old-policy"));
+        assert_eq!(
+            store.ann_cache.as_ref().map(|(generation, _)| *generation),
+            Some(active)
+        );
+        assert_eq!(store.start_or_resume("new-policy", &lock)?, (staging, true));
+        assert_eq!(
+            store.connection.query_row(
+                "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+                [staging],
+                |row| row.get::<_, usize>(0),
+            )?,
+            1
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn revision_race_cannot_publish_and_failed_activation_keeps_staging() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("notes.txt"), "source before change")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let p = lexical.passage_batch(0, 1)?.remove(0);
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        let (g, _) = store.start_or_resume("checked-policy", &lock)?;
+        let t = task(&p);
+        insert_complete(&mut store, g, &p, &t)?;
+        let old_revision = store.revision()?;
+        std::fs::write(root.join("notes.txt"), "source after change longer")?;
+        lexical.index_folder(&root)?;
+        assert!(store.activate_checked(g, old_revision).is_err());
+        assert_eq!(store.active_fingerprint()?, None);
+        assert_eq!(store.start_or_resume("checked-policy", &lock)?, (g, true));
+        assert_eq!(
+            store.connection.query_row(
+                "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
+                [g],
+                |r| r.get::<_, usize>(0)
+            )?,
+            1
+        );
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod completion_failure_tests {
+    use super::*;
+    use searchi_core::SearchIndex;
+
+    #[test]
+    fn failed_completion_rolls_back_chunk_membership_and_malformed_ranges_never_reuse() -> Result<()>
+    {
+        // SQLite durability fixture, not neural inference evidence.
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(root.join("notes.txt"), "UTF-8 source with 中文 preserved")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let p = lexical.passage_batch(0, 1)?.remove(0);
+        let t = SourceTask {
+            key: "durability-task".into(),
+            path: p.path.clone(),
+            source_type: "text".into(),
+            source_hash: content_hash(&p.text),
+            title_hash: content_hash(&p.filename),
+            policy: "exact-window-test".into(),
+            passage_id: p.id,
+        };
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        let (g, _) = store.start_or_resume("storage-geometry", &lock)?;
+        store.stage_task(g, &t)?;
+        store.insert(
+            g,
+            &p,
+            0,
+            &[TextWindow {
+                text: p.text.clone(),
+                start_byte: 0,
+                end_byte: p.text.len(),
+            }],
+            &[vec![1.0, 0.0]],
+        )?;
+        assert!(store.complete_source(g, &t, 2).is_err());
+        let (state,owned):(String,usize)=store.connection.query_row("SELECT state,(SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND task_key IS NOT NULL) FROM semantic_tasks WHERE generation=?1 AND task_key=?2",params![g,t.key],|r|Ok((r.get(0)?,r.get(1)?)))?;
+        assert_eq!(state, "running");
+        assert_eq!(owned, 0);
+        assert!(store.activate_checked(g, store.revision()?).is_err());
+        store.complete_source(g, &t, 1)?;
+        store.connection.execute("UPDATE semantic_chunks SET start_byte=1,snippet='different snippet' WHERE generation=?1",[g])?;
+        assert!(store.reuse_text_source(g, &t, &p).is_err());
+        assert!(store.active_fingerprint()?.is_none());
+        Ok(())
+    }
+}

fixtures/mixed/sources/docs__AUTOMATIC-INDEXING.md

@@ -0,0 +1,100 @@
+# Automatic folder updates
+
+The desktop now registers saved folders at startup and reconciles them in the
+background. Native notifications use pinned `notify` 8.2.0. They are hints:
+[notify documents](https://docs.rs/notify/8.2.0/notify/) missed events, network
+filesystem limitations and platform-specific editor behavior. Periodic scans
+remain necessary; a native watch failure keeps polling and reports the issue.
+
+## Scheduling and bounds
+
+- At most 64 roots and 128 retained paths per root; overflow becomes a full
+  reconciliation rather than an unbounded event queue.
+- A 750 ms quiet period coalesces bursts, with a five-second maximum delay.
+- Startup reconciles saved folders; ordinary verification runs every five minutes.
+- Busy manual indexing retains queued work. Failed scans retry with backoff from
+  five seconds to one minute. Events arriving during a scan remain pending.
+- Read/access-time events are ignored; native rescan flags still take priority.
+- Missing roots keep existing records and retry attachment. Successful
+  reattachment after a prior watch requests verification of missed media events.
+- Callback panics become a retryable failure. Shutdown signals cancellation,
+  wakes the owned thread and joins it without holding its state mutex.
+
+Scans own separate SQLite connections. They do not hold the foreground search
+mutex during traversal or extraction. A changed scan prepares its reader and
+briefly swaps it into the desktop. UI events refresh counts and the active query,
+reject results from before the change, and request normal semantic indexing.
+Refresh timing starts at the change event rather than include idle time since
+the original input. A bounded numeric snapshot retains the original input
+timings separately. Intended selection survives temporary result absence;
+explicit navigation changes that intent, and a new query or filter clears it.
+
+## Targeted reconciliation
+
+`SearchIndex::index_paths_with_options` handles bounded absolute normalized paths.
+Files update individually; directory hints reconcile only that subtree. Missing
+paths remove only records within their root and path boundary. Parent/child hints
+are coalesced. Outside paths, traversal and symlink ancestors are rejected.
+Read/traversal failure or cancellation rolls back the complete batch.
+
+The cancellable folder and path variants check between entries and before commit.
+An individual bounded parser input can finish before cancellation is observed.
+Parser process isolation and bounded shutdown time still require further work.
+
+Absolute exclusions complement directory-name exclusions. The desktop excludes
+its data, model and worker directories from both scans and notifications.
+Decoded video frames use owned scratch directories under application data,
+so temporary frames cannot feed back into the index when a home folder is watched.
+
+## Media freshness
+
+Core schema 13 adds `files.change_revision`, initially zero for existing records.
+Targeted media hints increment it even when file size and timestamp are unchanged,
+and remove old media tags. Full native-rescan/overflow hints take the targeted
+root path; ordinary periodic scans retain the metadata fast path for media.
+
+Semantic schema 8 records the matching `file_change_revision` on media chunks.
+An old media vector is ineligible after a hint until genuine byte-verified indexing
+completes. Unchanged media can reuse completed vectors with the current revision;
+changed bytes require new Google inference. Metadata probing captures the revision
+before work and writes only if it still matches, preventing a stale probe from
+restoring tags after a concurrent event. Generation activation remains guarded
+by the original corpus revision and actual source SHA validation.
+
+Periodic scans hash supported bounded document inputs, but do not hash every
+large media file. A same-stat media overwrite with no observed native hint remains
+a verification limitation; explicit semantic rebuilds verify actual media bytes.
+
+## Executed validation and remaining gates
+
+Windows native watcher tests observed real create/update/rename/delete events.
+The final Windows workspace run passed 123 tests (19 explicitly ignored), and
+the desktop suite passed 28 tests. Frontend build, 39 tests, formatting and strict
+workspace/desktop Clippy passed. The real CPU FFmpeg decoder regression passed
+with frame paths under the configured owned scratch parent.
+Coordinator tests cover overflow, rescan, busy retention, concurrent events,
+periodic scheduling, panic recovery and shutdown. Core tests cover same-stat
+documents/media, subtree and boundary deletion, exclusions, migrations, rollback,
+concurrent readers and cancellation after a write transaction starts.
+
+The real Google mixed-media CPU regression passed in 83.70 seconds. It verified
+that a same-stat image hint hides old vectors, changed image bytes cause one new
+image encode, unchanged hinted image/audio sources reuse vectors, and current
+revision tokens restore eligibility. This is correctness evidence on a tiny
+fixture, not relevance quality or search/indexing throughput evidence.
+
+Current-source native Linux checks also passed: 124 workspace tests, 28 desktop
+Release tests including actual inotify create/update/rename/delete, and the
+updated real Google mixed-media regression in 67.95 seconds. Native Linux
+interaction evidence is in
+[LINUX-DESKTOP-VALIDATION.md](LINUX-DESKTOP-VALIDATION.md). Actual watcher-driven
+UI change/create/rename/delete passed on `657de3d`; timing and selection issues
+found during those interactions require separate fixes and retesting.
+
+macOS native watcher execution, root disappearance/reappearance on
+real removable/network storage, native UI refresh interaction, representative
+large-corpus contention and power-loss behavior remain open. Full reconciliation
+still holds a write transaction during extraction and can contend with semantic
+writes; targeted batches reduce that work but do not establish a universal
+latency bound. Resource admission currently covers neural inference, not all
+filesystem, decoder, parser or ANN work.

fixtures/mixed/sources/docs__CODE-SEARCH.md

@@ -0,0 +1,80 @@
+# Search source code
+
+searchi searches source files already included in your selected indexed folders.
+In the desktop, open **All files** and select **Code** in the scope selector.
+Search an exact symbol or describe the implementation you want. The preview
+preserves whitespace and reports the passage's starting source line.
+Use `--code-only` in the CLI, or `code_only: true` in MCP, to restrict every
+retrieval lane to source-code files before candidate ranking. This is a file-type
+filter, not an access-control boundary or an instruction to index new folders.
+Other folder, extension, size and date filters still apply.
+
+Source support includes Rust, Python, JS/TS, C/C++, Go, Java/Kotlin, C#, Swift,
+Ruby, PHP, SQL, shells, PowerShell and additional common languages. Named
+`Makefile`, `GNUmakefile`, `Dockerfile`, `CMakeLists.txt` and `Justfile` are
+included. Ordinary JSON/TOML/YAML configuration remains searchable in All files.
+UTF-8 source passages span up to 128 lines with 32 lines of overlap to retain
+more routine context. This is bounded line-based extraction, not AST parsing.
+Binary/non-UTF-8 or oversized source remains filename-searchable.
+
+After upgrading an existing index, use **Reindex** to refresh source passages
+and rebuild semantic vectors. The schema migration schedules unchanged code
+for extraction again; it does not hide filenames while that work is pending.
+
+## Exact symbols and literal evidence
+
+An existing index supports names and source text without model assets:
+
+```sh
+searchi search --db /absolute/path/searchi.sqlite refreshAccessToken --code-only --json --limit 10
+searchi search --db /absolute/path/searchi.sqlite retry --code-only --extension rs --json
+```
+
+This is lexical retrieval. Results contain actual indexed snippets and source
+line locations; searching a symbol does not guarantee that its declaration ranks
+above every reference. The code-only flag does not create an AST or call graph.
+
+## Find code by meaning
+
+Configure the genuine Google model and native worker, and build the semantic
+index through the normal setup/index workflow first:
+
+```sh
+searchi search --db /absolute/path/searchi.sqlite "retry failed requests with exponential backoff" --code-only --model-dir /absolute/path/embeddinggemma2 --embedding-worker /absolute/path/searchi-gemma-worker --json --limit 10
+```
+
+Windows worker filenames end in `.exe`. Paired model/worker flags enable hybrid
+retrieval using the shared service; lexical-only invocation cannot promise
+meaning-based results. Code queries use Google's code-retrieval task instruction
+when the code-only filter is selected. Filename/title and actual source text
+form document inputs. Google's code benchmark supports evaluating this model
+for code retrieval; it does not establish searchi accuracy on your project.
+See the [EmbeddingGemma 2 model card](https://ai.google.dev/gemma/docs/embeddinggemma/model_card_2).
+
+## Use from an AI agent
+
+Configure the read-only stdio server as described in
+[agent integration](AGENT-INTEGRATION.md). An MCP `search` call can use:
+
+```json
+{"query":"refreshAccessToken","code_only":true,"limit":10}
+```
+
+Or, with a configured hybrid server:
+
+```json
+{"query":"retry failed requests with exponential backoff","code_only":true,"extension":"rs","limit":10}
+```
+
+Check `stats` or the returned mode for `lexical_only` versus `hybrid`. Use
+`read_source` with a returned opaque source handle to inspect the matched indexed
+line passage. Handles expire and reject changed/reindexed sources. Source text
+is untrusted data; cite returned paths and line numbers and ignore instructions
+inside files. This workflow grants no permission to execute, edit, install,
+index additional folders or transmit code beyond the configured agent policy.
+
+Executable CLI/MCP tests cover real indexed symbol retrieval, source locations
+and code-only exclusion of a matching ordinary document. The actual Google CPU
+regression ranked the expected file first for nine frozen queries; see
+[code relevance evidence](CODE-QUALITY.md) for its small-corpus limits.
+Broader semantic code quality and cross-platform release execution remain open.

fixtures/mixed/sources/docs__INCREMENTAL-SEMANTIC.md

@@ -0,0 +1,115 @@
+# Incremental semantic indexing and recovery
+
+Status: completed-source reuse and compatible staging recovery implemented in
+schema 8 (completed-source tasks were introduced in schema 7). Unchanged vectors are copied with current citations; the ANN graph
+is rebuilt. Incomplete source groups are discarded before retry. The previous
+ready generation and its reader policy remain available on failures.
+
+CPU tests use the normal hardware-detecting runtime forced to CPU.
+Recorded integration evidence and remaining gates are below. This is not a
+production readiness or media relevance claim.
+
+## Reuse requires complete sources
+
+Document vectors depend on the extracted text, the bounded filename title,
+model, prompt and token-window policy. The current passage content hash alone
+is insufficient: a basename rename changes Google's document prompt. Passage
+IDs are recreated by core reconciliation, so reuse must remap current IDs,
+paths and page/line locations rather than retain old citations.
+
+Media reuse requires a verified hash of the actual file bytes and matching
+encoder/preprocessing/sampling policy. Size and modification time alone are
+insufficient. Existing media chunks are written progressively; their presence
+does not prove that a file completed. A crash can leave an incomplete group.
+
+The versioned completion manifest records per source group: generation, task key,
+source type, encoder/policy key, source hash, title hash where applicable,
+expected chunk count and completion state. Associate chunks with their group.
+Only completed groups can supply reusable vectors. Historical generations
+without completion evidence remain searchable but require one rebuild before
+their vectors qualify for reuse.
+
+Start with matching complete-generation policies. Component encoder keys can
+later let unchanged document vectors survive an audio-sampling policy change.
+Do not silently omit image, audio or video work to make a rebuild cheaper.
+
+## Durable work and atomic publication
+
+Persist bounded pending/running/done/failed tasks in SQLite. Tasks retain their
+source identity, attempts and a bounded error. Obtain the existing OS build
+lock before starting or resuming a compatible staging generation; only then
+can interrupted running tasks be reset. Keep inference outside write
+transactions. Commit source completion, verified chunk membership and task
+state atomically. Incomplete groups must be discarded or resumed without
+qualifying as completed reuse sources.
+
+Discover work in bounded keyset batches. Before publication, reconcile staging
+against current indexed sources: remove deleted sources, invalidate changed
+identities and enqueue additions. Verify exact document window slices and
+remap citations. For media, confirm the file is unchanged after extraction and
+inference. Retain the 100,000-chunk budget and explicit per-file media failures.
+
+Build the ANN graph only from eligible completed groups. Publish under a
+transaction that checks the reconciled corpus revision; a concurrent source
+change must trigger reconciliation/retry. Storage or activation failure must
+preserve both the previous active generation and the same reader's policy.
+
+## Acceptance gates
+
+- [x] Schema-6 storage migration preserves active-generation search; legacy
+  chunks without completion evidence cannot supply reusable vectors.
+- [x] Rebuild mixed text/code/image/audio/video fixtures unchanged with zero
+  document/media inference calls and identical vectors and source locations.
+- [x] Change one document and one image; re-encode only affected groups.
+  Image regression preserves file size and exact mtime to exercise SHA identity.
+- [x] Re-encode text after a basename rename; reuse a folder-only move with
+  current citations.
+- [ ] Verify media bytes and current citations after a media-file rename.
+- [ ] Regenerate affected media when sampling policy changes; retain eligible
+  text groups when component encoder-key reuse is implemented.
+- [ ] Terminate after a committed batch and resume without repeating completed
+  inference. Never reuse or publish an unfinished source group.
+- [x] Same-size/mtime text mutation during genuine inference rejects publication;
+  a mutation before rebuilding fails before text encoding and ordinary rescans
+  reextract it. Unaffected vectors survive resumed retry.
+- [ ] Concurrent deletion/media mutation and comprehensive queued/running source
+  search freshness tests.
+- [x] Inject activation failure, preserve existing results and resume completed
+  staging with zero document inference; cleanup-failure unit regression retains
+  the primary error and old reader policy.
+- [x] Use genuine Google CPU inference for reuse integration checks. Storage
+  fixtures alone do not establish retrieval quality or reuse equivalence.
+- [ ] Measure preparation, the first foreground query and warm p95 while
+  resumed indexing runs; record actual inference/reuse counts and resource use.
+
+Foreground query priority and bounded persistent workers remain part of the
+first-use requirement. Faster unchanged rebuilds alone do not prove responsive
+search while indexing.
+
+## Current limits
+
+Reuse requires the complete generation policy to match. Component-level reuse
+across media sampling changes remains open. Durable staging resumes at completed
+source boundaries; a real process-kill test during a multi-batch source remains
+open. Storage tests cover discarded partial groups and corpus revision races.
+
+Full source hashes are checked after ANN construction and serialization, before
+opening the publication write transaction. Indexed changes are also rejected by
+the transaction's revision guard. This narrows the filesystem race; it does not
+make file reads and SQLite publication one atomic filesystem operation.
+
+Core schema 12 records the SHA-256 of the exact bounded bytes consumed by each
+parser. Semantic builds verify that provenance before reusing or encoding text.
+Same-size/mtime changes are detected by normal folder rescanning. Legacy records
+retain lexical search but need a rescan before semantic rebuilding or serving
+agent excerpts. Excerpts reject missing or mismatched source provenance.
+
+Bounded discovery now prunes obsolete resumed groups before inference. Compatible
+renamed staging groups survive with current memberships through a transactional,
+disk-backed SQLite snapshot; active vectors are untouched. Genuine duplicate
+current groups still count toward the 100,000-chunk ceiling. Hashing, discovery,
+media metadata probes, snapshot copying and graph rebuilding have costs: zero
+encoder calls does not mean zero work or prove fast first input.
+
+Synthetic colored-image and sine-wave fixtures prove actual vector reuse and
+citation equality; they do not establish useful media relevance.

fixtures/mixed/sources/docs__PERFORMANCE.md

@@ -0,0 +1,219 @@
+# Search performance diagnostics
+
+Search performance must distinguish total request latency from the work inside
+it. The shared Rust schema exposes basic metrics for routine searches and a
+bounded nested trace for deliberate debugging. It uses monotonic `Instant`
+timing, not wall-clock timestamps. No query text, source passages or file paths
+belong in stage names/counters.
+
+## Shared response
+
+`SearchResponse<T>` contains `results: Vec<T>` and `metrics: SearchMetrics`.
+
+| Field | Meaning |
+| --- | --- |
+| total_ms | Elapsed time inside the collector's measured request boundary |
+| result_count | Number of final returned results |
+| trace | null in basic mode; debug spans in begin order |
+| dropped_spans | Spans omitted when the 512-span ceiling or invalid parent is encountered |
+
+A `SearchSpan` has a static `name`, relative `start_ms`, `duration_ms`, optional
+parent index and integer `counters`. Parent indices reference earlier spans in
+the same trace; child timings lie within their parent. Nested durations overlap,
+so summing every span would double-count work. Counters are capped at 32 keys per
+span, with 96-character labels. Missing spans due to the cap are visible through
+`dropped_spans`, never presented as complete instrumentation.
+
+Basic mode retains only the request start Instant and result count; there is no
+per-stage span Vec or diagnostic string allocation. Debug mode captures up to
+512 spans. Scope timings preserve normal closure return values and errors.
+Already-ended or foreign tokens are harmless, and unfinished scopes close at
+collector finish. Closing a parent closes its open descendants for containment.
+
+## Instrumenting the shared engine
+
+```rust
+use searchi_core::metrics::{MetricsMode, TraceCollector};
+let mut trace = TraceCollector::new(MetricsMode::Debug);
+let results = trace.timed("search", None, |trace, parent| {
+    trace.timed("candidate_selection", parent, |trace, token| {
+        trace.counter(token, "candidate_count", candidates.len() as u64);
+        retrieve_candidates(candidates)
+    })
+});
+let metrics = trace.finish(results.len());
+```
+
+This is an API illustration, not a measured benchmark. `begin(name,parent)` /
+`end(token,counters)` supports explicit scopes spanning locks or separate calls;
+`SpanToken` is Copy and exposes its index. `counter` records bounded integer
+counts without source data. `elapsed_ms` reports elapsed request time. Capture
+queue/lock boundaries in the caller, then pass that same collector into the
+shared core/service so stages refer to one origin.
+
+Useful actual boundaries include request admission/reader lock, filter snapshot,
+filename/FTS retrieval, tokenizer/query inference, active ANN load, eligibility
+refresh, filtered HNSW traversal, per-file deduplication, ranking/fusion and
+result assembly. Name only work that is truly measured. Do not label the whole
+worker round trip as pure neural execution or fabricate stage splits from a
+single elapsed number. Model load/warmup and initial eligibility work should be
+reported separately from a retained worker's first foreground and warm queries.
+
+## CLI and desktop integration
+
+The CLI exposes these actual modes:
+
+```sh
+searchi search --db searchi.sqlite contract --json
+searchi search --db searchi.sqlite contract --performance --json
+searchi search --db searchi.sqlite contract --debug-performance --json
+```
+
+Plain `--json` preserves the legacy result array. `--performance --json` emits
+`SearchResponse` with basic metrics; `--debug-performance --json` emits the
+same envelope with detailed spans. Normal terminal output prints results and
+reports elapsed time/count on stderr, including CLI index/session initialization.
+Debug terminal mode also prints the detailed metrics on stderr. Legacy Vec
+interfaces remain wrappers for compatibility. The desktop should use a compact ordinary summary and an optional
+diagnostic view, without confusing score/similarity with elapsed time or confidence.
+
+Both measured GUI commands start the native clock at command entry, include instrumented queue/lock waits, and finish before response serialization. UI timing separately captures input/debounce, lane/retry waits, invoke/round trip and synchronous DOM updates. Display latency ends at the DOM update; actual compositor paint and deferred media preview loading are not measured. Their overlapping boundaries must not be summed. Native metrics exclude frontend delays and post-return paint. A fast Rust request does not
+prove a responsive first foreground interaction. Conversely, pending setup or
+model loading must not be silently included in a warm-query claim.
+
+## Reproducible performance validation
+
+Automatic filesystem refreshes start a new frontend measurement at the refresh
+event. `trigger: "filesystem_refresh"` distinguishes them from user searches;
+`SHOWN` and `latest_displayed_ms` exclude time spent idle before that event.
+`original_input` retains a fixed-size snapshot of the original debounce and
+display timings. Refreshes do not replay previous attempts or append ancestor
+traces. A refresh before the first response leaves that original response timing
+null rather than inventing a first-input measurement.
+
+Record build profile, CPU/device and thread budget, VM/host resources, corpus and
+active generation, filters, cold/warm state, concurrent indexing and setup costs.
+Use multiple queries and p50/p95 distributions; retain raw reports. Compare basic
+and debug mode on the same corpus to quantify instrumentation overhead. Debug
+traces guide investigation but are not a substitute for representative benchmarks.
+
+Current recorded filename, actual Google seed and synthetic ANN evidence is in
+[VALIDATION](VALIDATION.md) and [SEMANTIC](SEMANTIC.md). Media quality/timing is in
+[MEDIA-QUALITY](MEDIA-QUALITY.md). Synthetic vectors measure index behavior only;
+real Google inference remains required for product semantic search. Platform,
+large-corpus, concurrent-indexing and GPU performance gates remain independent.
+## Worker round-trip diagnostics
+
+The retained hybrid service now keeps a bounded 32-query vector LRU. Debug
+`semantic.query_embedding` spans report `query_cache_hit` and `query_cache_miss`.
+A hit reuses an actual model vector and performs no neural inference; it must
+not carry the previous worker's tokenization/inference counters. Filters, current
+source eligibility, ANN retrieval and fusion still execute. Warmup performs an
+actual neural probe and does not seed an unseen foreground-query result.
+Compare fresh queries, repeats and filter changes separately. The current
+[EmbeddingGemma 2 audit](EMBEDDINGGEMMA2-AUDIT.md) includes paired real-CPU
+measurements, startup boundaries and remaining indexing/media costs.
+
+Measured desktop failures return `{message, metrics}` and retain the trace for
+the failed attempt. Busy retries are separate attempts. If an unexpected worker
+task failure loses its collector, elapsed time is retained, the trace is null,
+and a dropped-span marker makes the missing detail explicit. Error messages are
+excluded from the exported timing report.
+
+Embedding IPC spans expose named `*_us` integer counters for request encoding,
+writing, worker wait/response decode, validation, and native tokenization, batch
+preparation, inference and pooling/normalization when genuinely reported. They
+also expose `worker_token_count` and the GPU-to-CPU restart count. Microsecond
+counters are durations, not occurrence counts; their suffix records units.
+
+The native worker clock is not aligned with Rust's collector origin. Native
+stage values remain counters on the IPC span rather than invented positioned
+child spans. Worker wait includes response-reader JSON parsing and scheduling;
+it is not pure inference time. The encompassing round trip and its component
+counters overlap. A fallback/restart may include setup outside the last worker's
+reported neural pass; retain that distinction when diagnosing slow requests.
+
+## Small instrumentation-overhead diagnostic
+
+The deterministic example creates 1,000 local text files, builds/prewarms SQLite,
+then alternates basic/debug order over 24 exact filename queries and 20 repeats.
+Both modes must retrieve the correct file. It records 480 samples per mode,
+outer-call p50/p95/mean and actual debug span counts, excluding fixture setup
+and JSON serialization. It runs no model, media decoder or network.
+
+```powershell
+cargo run --release -p searchi-core --example trace_overhead -- .searchi/trace-overhead.json
+```
+
+This tiny warm lexical corpus measures diagnostic overhead under the current
+VM conditions. It does not certify production search latency or model-query
+tracing overhead. Keep the raw report and profile/device/concurrency caveats;
+repeat on representative corpora and compare instrumentation policies before
+claiming low overhead across every retrieval mode.
+
+Windows CPU release diagnostic, 2026-10-12 (480 samples per mode):
+
+| Mode | Mean | p50 | p95 |
+| --- | --- | --- | --- |
+| Basic | 0.556 ms | 0.440 ms | 0.934 ms |
+| Detailed | 0.968 ms | 0.819 ms | 1.615 ms |
+
+The detailed searches retained 14 spans each. These measurements include the
+search itself and metrics assembly, and exclude fixture setup and JSON encoding.
+Other work in the VM could contend for CPU. [Raw samples and measurement scope](evidence/windows-cpu-trace-overhead.json).
+The recorded Windows x86-64 release run retained 480 raw samples per mode and
+14 actual debug spans per request:
+
+| Mode | Mean | p50 | p95 |
+| --- | ---: | ---: | ---: |
+| Basic | 0.556 ms | 0.440 ms | 0.934 ms |
+| Debug | 0.968 ms | 0.819 ms | 1.615 ms |
+
+[Raw samples and measurement contract](evidence/windows-cpu-trace-overhead.json)
+include the active VM-contention caveat: native photo inference and other work
+may overlap. This is an absolute small-corpus comparison, not an old-baseline
+regression claim or proof of model/media tracing overhead. The extra cost
+supports keeping detailed mode optional rather than claiming it is free.
+
+## Prepared-reader publication
+
+The desktop shares a bounded process-local prepared ANN between indexing and
+its retained query service. The graph is registered before SQL activation and
+matched by canonical database, dimensions, generation, policy and checksum.
+`semantic.prepared_graph_lookup` measures that handoff; a successful handoff
+sets `semantic.ann_load` cache_hit to one, without blob read/checksum/deserialization.
+The query model and real-vector LRU remain warm. Eligibility refresh and query
+inference still have costs and are measured independently.
+
+The actual CPU regression checks both a repeated cached query and an unseen
+query after publication; the unseen input must record query_cache_hit zero.
+See [RECOVERY-AND-HANDOFF.md](RECOVERY-AND-HANDOFF.md). Its single-document
+request times do not establish representative-corpus p95, native first input
+or foreground priority during indexing.
+
+## Shared inference admission
+
+The desktop now shares one CPU thread budget across its persistent query and
+indexing workers. The default is two threads, clamped to available logical CPUs
+and a maximum of eight. Indexing uses one thread; the query worker uses the
+remaining threads (at least one). Pending searches have priority over new
+background encodes, and document indexing yields between individual encodes.
+Cached query vectors do not acquire an inference permit.
+
+Settings version two stores the total budget. Version-one settings counted
+threads per worker; migration adds one background thread, capped at eight.
+An explicit version-two budget of one remains one. Resource changes are rejected
+during active semantic work; a failed settings save preserves the warm session.
+
+Debug traces record `semantic.inference.admission_wait` separately from native
+inference. Admission waits are bounded at 30 seconds. Already executing native
+inference cannot be preempted: a one-thread budget or single-core machine can
+still delay a search during a long media encode. This budget covers admitted
+inference worker threads, not extraction, decoding, ANN construction, rendering
+or total operating-system CPU usage.
+
+The real Google CPU regression `budget_google_cpu` confirmed overlapping query
+and background inference with a two-thread budget: one query thread, one indexing
+thread, six background document calls, an unseen query taking 103.407 ms including
+0.034 ms admission in the final scoped-cleanup run. These are Windows VM debug measurements on a small fixture,
+not representative-corpus p95, native UI latency, GPU or scale evidence.

tools/mixed-eval/README.md

@@ -0,0 +1,35 @@
+# Mixed retrieval evaluator
+
+`evaluate.py` verifies the frozen labels and every asset, copies only those
+assets into an owned temporary corpus, and indexes it with the actual CLI.
+Semantic indexing uses the real shared service and explicit Google worker/model
+and CPU FFmpeg paths. Querying uses the same public read-only MCP search tool
+that local agents use. Lexical and hybrid modes are reported separately; code
+queries compare All files against Code scope. A semantic fallback is an error,
+not a lexical result mislabeled as model evidence. Model files are not downloaded.
+
+```powershell
+$env:SEARCHI_FORCE_CPU = '1'
+python tools/mixed-eval/evaluate.py --verify-only
+python -m unittest discover -s tools/mixed-eval -p 'test_*.py'
+python tools/mixed-eval/evaluate.py --cli target/release/searchi.exe --models .searchi/models/embeddinggemma2 --worker .searchi/runtime/auto/Release/searchi-gemma-worker.exe --ffmpeg .searchi/media-runtime/bin/ffmpeg.exe --ffprobe .searchi/media-runtime/bin/ffprobe.exe --output .searchi/qa/mixed-v1.json
+```
+
+Use actual absolute runtime paths for your platform if these development paths
+differ. Inference runs must be coordinated with other CPU measurements. Expect
+media indexing to take minutes; its explicit one-hour bound is a safety ceiling,
+not a performance target. Indexing and query errors remain visible. Temporary
+corpora/indexes are cleaned after the owned sessions close. Partial query results
+are retained in the requested output if a later request fails.
+
+Reported metrics are file-level top1, reciprocal rank and recall@5 after file
+identity deduplication. Hard negatives in the first five and no-answer queries
+returning any result are reported explicitly. This does **not** calibrate a
+rejection threshold. Timestamp labels are retained but scored only for file
+identity: use the separate media evaluator for segment localization. Unsupported
+temporal filters are recorded as skipped, never counted as passed.
+
+The raw report retains queries, expected IDs, ordered hits, source locations,
+scores, modes/scopes/languages and MCP roundtrip times. Those times exclude
+process/model initialization and do not represent native UI latency. Keep raw
+reports under ignored `.searchi`; review deliberate evidence before publishing.

tools/mixed-eval/evaluate.py

@@ -0,0 +1,156 @@
+"""Frozen mixed diagnostic via the installed CLI's shared engine and MCP API."""
+import argparse
+from datetime import datetime, timezone
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+
+ROOT = Path(__file__).resolve().parents[2]
+FIXTURE = ROOT / 'fixtures/mixed'
+
+
+def digest(path):
+    with path.open('rb') as stream:
+        return hashlib.file_digest(stream, 'sha256').hexdigest()
+
+
+def load_fixture(directory=FIXTURE):
+    raw = (directory / 'fixture.json').read_bytes()
+    lock = json.loads((directory / 'labels.lock.json').read_bytes())
+    if hashlib.sha256(raw).hexdigest() != lock['fixture_sha256']:
+        raise ValueError('Frozen labels changed; create a new version, never refreeze v1')
+    fixture = json.loads(raw)
+    allowed = directory.parent.resolve()
+    ids, names = set(), set()
+    for asset in fixture['assets']:
+        path = (directory / asset['path']).resolve()
+        if not path.is_relative_to(allowed) or not path.is_file():
+            raise ValueError('Asset must be an existing file below fixtures')
+        if asset['id'] in ids or path.name in names:
+            raise ValueError('Asset IDs and materialized filenames must be unique')
+        ids.add(asset['id']); names.add(path.name)
+        if path.stat().st_size != asset['bytes'] or digest(path) != asset['sha256']:
+            raise ValueError('Frozen asset changed: ' + asset['id'])
+    for query in fixture['queries']:
+        # Temporal/media labels can include structured entries; normalize below.
+        if any(identifier not in ids for identifier in expected_ids(query)):
+            raise ValueError('Unknown expected asset')
+    return fixture, lock
+
+
+def expected_ids(query):
+    return {item if isinstance(item, str) else item['asset'] for item in query['expected']}
+
+
+def score(query, hits):
+    expected = expected_ids(query)
+    unique = list(dict.fromkeys(hit['asset'] for hit in hits))
+    ranks = [i + 1 for i, asset in enumerate(unique) if asset in expected]
+    first = min(ranks, default=None)
+    return {'first_relevant_rank': first,
+            'top1': first == 1 if expected else None,
+            'recall_at_5': len(expected.intersection(unique[:5])) / len(expected) if expected else None,
+            'reciprocal_rank': 1 / first if first else 0 if expected else None,
+            'no_answer_returned_any': bool(hits) if not expected else None,
+            'hard_negative_in_top5': any(asset in query.get('hard_negatives', []) for asset in unique[:5])}
+
+
+def summarize(rows):
+    groups = {}
+    for row in rows:
+        if 'unsupported' in row:
+            continue
+        key = (row['mode'], row['scope'], row['query']['category'], row['query']['language'])
+        groups.setdefault(key, []).append(row['score'])
+    result = []
+    for key, values in sorted(groups.items()):
+        group = dict(zip(('mode', 'scope', 'category', 'language'), key))
+        group['queries'] = len(values)
+        for metric in ('top1', 'recall_at_5', 'reciprocal_rank', 'no_answer_returned_any', 'hard_negative_in_top5'):
+            numbers = [v[metric] for v in values if v[metric] is not None]
+            group[metric] = sum(numbers) / len(numbers) if numbers else None
+        result.append(group)
+    return result
+
+
+def run(args):
+    if os.environ.get('SEARCHI_FORCE_CPU') != '1':
+        raise ValueError('Set SEARCHI_FORCE_CPU=1; this evaluator must never run GPU inference')
+    fixture, lock = load_fixture()
+    manifest_path = ROOT / 'tools/embeddinggemma2/model-manifest.json'
+    manifest = json.loads(manifest_path.read_bytes())
+    for model in manifest['files']:
+        path = args.models / model['path']
+        if path.stat().st_size != model['bytes'] or digest(path) != model['sha256']:
+            raise ValueError('Model bytes differ from the pinned Google manifest')
+    spec = importlib.util.spec_from_file_location('searchi_benchmark', ROOT / 'tools/embeddinggemma2/benchmark.py')
+    benchmark = importlib.util.module_from_spec(spec); spec.loader.exec_module(benchmark)
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    rows = []
+    with tempfile.TemporaryDirectory(prefix='searchi-mixed-') as temporary:
+        base = Path(temporary); corpus = base / 'corpus'; corpus.mkdir(); database = base / 'index.sqlite'
+        names = {}
+        for asset in fixture['assets']:
+            source = (FIXTURE / asset['path']).resolve()
+            shutil.copyfile(source, corpus / source.name)
+            names[source.name] = asset['id']
+        subprocess.run([str(args.cli), 'index', '--db', str(database), '--root', str(corpus)], check=True, timeout=300)
+        subprocess.run([str(args.cli), 'embed', '--db', str(database), '--model-dir', str(args.models),
+                        '--embedding-worker', str(args.worker), '--cpu-threads', '1',
+                        '--ffmpeg', str(args.ffmpeg), '--ffprobe', str(args.ffprobe)], check=True, timeout=3600)
+        for mode in ('lexical_only', 'hybrid'):
+            command = [str(args.cli), 'mcp', '--db', str(database)]
+            if mode == 'hybrid':
+                command += ['--model-dir', str(args.models), '--embedding-worker', str(args.worker), '--cpu-threads', '1']
+            session = benchmark.Session(command)
+            try:
+                session.mcp(1, 'initialize', {'protocolVersion': '2025-03-26', 'capabilities': {}, 'clientInfo': {'name': 'mixed-eval', 'version': '1'}})
+                session.send({'jsonrpc': '2.0', 'method': 'notifications/initialized'})
+                request = 2
+                for query in fixture['queries']:
+                    scopes = ('all_files', 'code') if query['category'] == 'code' else ('all_files',)
+                    for scope in scopes:
+                        options = {'query': query['text'], 'limit': 20, 'code_only': scope == 'code'}
+                        filters = query.get('filters', {})
+                        if 'time_range_s' in filters:
+                            rows.append({'mode': mode, 'scope': scope, 'query': query, 'unsupported': 'Temporal-range search filter is not supported'})
+                            continue
+                        if 'extension' in filters:
+                            options['extension'] = filters['extension']
+                        response, elapsed = session.mcp(request, 'tools/call', {'name': 'search', 'arguments': options}); request += 1
+                        if response.get('isError'):
+                            raise RuntimeError('Shared search failed: ' + json.dumps(response))
+                        content = response['structuredContent']
+                        if content['mode'] != mode:
+                            raise RuntimeError('Unexpected semantic fallback; not valid quality evidence')
+                        hits = [{'asset': names[hit['filename']], 'score': hit['score'], 'match_kind': hit['match_kind'], 'source': hit.get('source')} for hit in content['results']]
+                        rows.append({'mode': mode, 'scope': scope, 'query': query, 'hits': hits, 'score': score(query, hits), 'roundtrip_ms': elapsed})
+                        # Retain partial actual evidence if a later request fails.
+                        args.output.write_text(json.dumps({'complete': False, 'rows': rows}, indent=2, ensure_ascii=False), encoding='utf-8')
+            finally:
+                session.close()
+    report = {'complete': True, 'recorded_at': datetime.now(timezone.utc).isoformat(), 'fixture_sha256': lock['fixture_sha256'],
+              'model_manifest_sha256': digest(manifest_path), 'model': manifest,
+              'device': 'cpu', 'cpu_threads': 1, 'cli_sha256': digest(args.cli), 'worker_sha256': digest(args.worker),
+              'scope_note': 'Small mixed diagnostic, reused previously tested media labels; file-level relevance only. No temporal-localization or production-scale claims. MCP roundtrip excludes initialization and is not UI latency.',
+              'rows': rows, 'summary': summarize(rows)}
+    args.output.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding='utf-8')
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--verify-only', action='store_true')
+    for flag in ('cli', 'models', 'worker', 'ffmpeg', 'ffprobe', 'output'):
+        parser.add_argument('--' + flag, type=Path)
+    args = parser.parse_args()
+    if args.verify_only:
+        fixture, lock = load_fixture(); print('Verified', len(fixture['assets']), 'assets and', len(fixture['queries']), 'frozen queries', lock['fixture_sha256'])
+    else:
+        if any(getattr(args, flag) is None for flag in ('cli', 'models', 'worker', 'ffmpeg', 'ffprobe', 'output')):
+            parser.error('Supply all explicit runtime paths and output, or use --verify-only')
+        run(args)

tools/mixed-eval/test_evaluate.py

@@ -0,0 +1,57 @@
+"""Meaningful scorer and frozen-provenance checks; no inference."""
+import importlib.util
+from pathlib import Path
+import tempfile
+import unittest
+
+spec = importlib.util.spec_from_file_location('mixed_eval', Path(__file__).with_name('evaluate.py'))
+evaluate = importlib.util.module_from_spec(spec); spec.loader.exec_module(evaluate)
+
+
+class EvaluationTests(unittest.TestCase):
+    def test_frozen_real_assets_verify(self):
+        fixture, _ = evaluate.load_fixture()
+        self.assertEqual(len(fixture['assets']), 32)
+        self.assertEqual(len(fixture['queries']), 91)
+        self.assertTrue(any(a.get('source_repository') for a in fixture['assets']))
+
+    def test_multiple_relevant_files_and_duplicate_hits(self):
+        query = {'expected': ['a', 'b'], 'hard_negatives': ['x']}
+        result = evaluate.score(query, [{'asset': x} for x in ['x', 'a', 'a', 'c', 'b']])
+        self.assertEqual(result['first_relevant_rank'], 2)
+        self.assertEqual(result['recall_at_5'], 1)
+        self.assertEqual(result['reciprocal_rank'], .5)
+        self.assertTrue(result['hard_negative_in_top5'])
+
+    def test_no_answer_is_not_counted_as_positive_recall(self):
+        result = evaluate.score({'expected': []}, [{'asset': 'unrelated'}])
+        self.assertTrue(result['no_answer_returned_any'])
+        self.assertIsNone(result['recall_at_5'])
+        self.assertIsNone(result['top1'])
+        self.assertIsNone(result['reciprocal_rank'])
+
+    def test_temporal_labels_file_relevance_only(self):
+        result = evaluate.score({'expected': [{'asset': 'speech', 'start_s': 20, 'end_s': 25}]}, [{'asset': 'speech', 'source': {'number': 0}}])
+        self.assertTrue(result['top1'])  # Deliberately no invented timestamp success metric.
+
+    def test_label_tampering_rejected_before_assets(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory)
+            (path / 'fixture.json').write_text('{"assets": []}')
+            (path / 'labels.lock.json').write_text('{"fixture_sha256": "wrong"}')
+            with self.assertRaisesRegex(ValueError, 'Frozen labels changed'):
+                evaluate.load_fixture(path)
+
+    def test_summary_separates_scope_and_language(self):
+        rows = []
+        for scope, lang, expected, found in [('all_files', 'en', ['a'], ['b']), ('code', 'en', ['a'], ['a']), ('all_files', 'de', [], ['b'])]:
+            query = {'expected': expected, 'language': lang, 'category': 'code'}
+            rows.append({'mode': 'hybrid', 'scope': scope, 'query': query, 'score': evaluate.score(query, [{'asset': a} for a in found])})
+        groups = evaluate.summarize(rows)
+        self.assertEqual(len(groups), 3)
+        self.assertEqual(next(g for g in groups if g['scope'] == 'code')['top1'], 1)
+        self.assertEqual(next(g for g in groups if g['language'] == 'de')['no_answer_returned_any'], 1)
+
+
+if __name__ == '__main__':
+    unittest.main()