Use measured query-only context across desktop CLI and MCP

This commit is contained in:
karl committed 2026-10-13 07:30:40 +02:00
1 parent 95c72c5fd4
commit 88cd9c7f5a
12 files changed
+582 -20

No files matched your search

+1 -1
View File
@@ -203,7 +203,7 @@ fn main() -> Result<()> {
});
let results = if let (Some(model), Some(runtime)) = (model_dir, embedding_worker) {
let mut service = trace.timed("cli.session_initialize", None, |_, _| {
HybridSearch::open(&db, model, runtime, usize::from(cpu_threads))
HybridSearch::open_query(&db, model, runtime, usize::from(cpu_threads))
})?;
service
.search_traced(&query, usize::from(limit), &options, &mut trace, None)
+1 -1
View File
@@ -450,7 +450,7 @@ pub fn serve(
"No semantic generation; run explicit indexing/embedding outside MCP first"
);
let mut hybrid = HybridSearch::open(&db, models, worker, threads)?;
let mut hybrid = HybridSearch::open_query(&db, models, worker, threads)?;
hybrid.warmup()?;
+123 -3
View File
@@ -19,9 +19,22 @@ impl Drop for Server {
}
impl Server {
fn start(db: &std::path::Path) -> Self {
let mut child = Command::new(env!("CARGO_BIN_EXE_searchi"))
.args(["mcp", "--db"])
.arg(db)
Self::start_with_assets(db, None)
}
fn start_with_assets(
db: &std::path::Path,
assets: Option<(&std::ffi::OsStr, &std::ffi::OsStr)>,
) -> Self {
let mut command = Command::new(env!("CARGO_BIN_EXE_searchi"));
command.args(["mcp", "--db"]).arg(db);
if let Some((model, worker)) = assets {
command
.arg("--model-dir")
.arg(model)
.arg("--embedding-worker")
.arg(worker);
}
let mut child = command
.env("SEARCHI_FORCE_CPU", "1")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
@@ -233,3 +246,110 @@ fn missing_database_never_creates_file_or_protocol_noise() {
assert!(output.stdout.is_empty());
assert!(!db.exists());
}
#[test]
#[ignore = "Requires verified Google model and current CPU-capable native worker"]
fn actual_google_cli_and_mcp_use_query_role_with_cited_results() {
let models = std::env::var_os("SEARCHI_GEMMA_MODEL_DIR").expect("Set models");
let worker = std::env::var_os("SEARCHI_GEMMA_WORKER").expect("Set worker");
let fixture = tempfile::tempdir().unwrap();
let folder = fixture.path().join("documents");
std::fs::create_dir(&folder).unwrap();
std::fs::write(
folder.join("supplier.txt"),
"The supplier agreement may be terminated with sixty days written notice.",
)
.unwrap();
std::fs::write(
folder.join("garden.txt"),
"Vegetable beds and rainwater collection in the garden.",
)
.unwrap();
std::fs::write(
folder.join("retry.rs"),
"pub fn retry_backoff(attempt: u32) -> u64 {\n 100 * 2_u64.pow(attempt)\n}\n",
)
.unwrap();
let db = fixture.path().join("index.sqlite");
searchi_core::SearchIndex::open(&db)
.unwrap()
.index_folder(&folder)
.unwrap();
let indexer = searchi_service::HybridSearch::open(&db, &models, &worker, 1).unwrap();
assert_eq!(indexer.worker_info().unwrap().role, "indexing");
drop(indexer);
let embedded = Command::new(env!("CARGO_BIN_EXE_searchi"))
.args(["embed", "--db"])
.arg(&db)
.arg("--model-dir")
.arg(&models)
.arg("--embedding-worker")
.arg(&worker)
.arg("--text-only")
.env("SEARCHI_FORCE_CPU", "1")
.output()
.unwrap();
assert!(
embedded.status.success(),
"{}",
String::from_utf8_lossy(&embedded.stderr)
);
let output = Command::new(env!("CARGO_BIN_EXE_searchi"))
.args(["search", "--db"])
.arg(&db)
.args(["cancel the supplier partnership", "--json"])
.arg("--model-dir")
.arg(&models)
.arg("--embedding-worker")
.arg(&worker)
.env("SEARCHI_FORCE_CPU", "1")
.output()
.unwrap();
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
let cli: Value = serde_json::from_slice(&output.stdout).unwrap();
assert_eq!(cli[0]["filename"], "supplier.txt");
let mut server = Server::start_with_assets(&db, Some((&models, &worker)));
server.request(1,"initialize",json!({"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"query-role-regression","version":"1"}}));
server.send(json!({"jsonrpc":"2.0","method":"notifications/initialized"}));
let stats = server.request(2, "tools/call", json!({"name":"stats","arguments":{}}));
let status = &stats["result"]["structuredContent"];
assert_eq!(status["worker"]["role"], "query");
assert_eq!(status["worker"]["context_tokens"], 512);
assert_eq!(status["worker"]["device"], "cpu");
assert_eq!(status["worker"]["media_supported"], true);
let found = server.request(
3,
"tools/call",
json!({"name":"search","arguments":{"query":"cancel the supplier partnership","limit":2}}),
);
let first = &found["result"]["structuredContent"]["results"][0];
assert_eq!(first["filename"], "supplier.txt");
let handle = first["source_handle"].as_str().unwrap();
let source = server.request(
4,
"tools/call",
json!({"name":"read_source","arguments":{"handle":handle}}),
);
assert!(source["result"]["structuredContent"]["text"]
.as_str()
.unwrap()
.contains("sixty days"));
let code=server.request(5,"tools/call",json!({"name":"search","arguments":{"query":"retry failed requests with exponential backoff","code_only":true}}));
let first = &code["result"]["structuredContent"]["results"][0];
assert_eq!(first["filename"], "retry.rs");
assert_eq!(first["source"]["kind"], "line");
let handle = first["source_handle"].as_str().unwrap();
let source = server.request(
6,
"tools/call",
json!({"name":"read_source","arguments":{"handle":handle}}),
);
assert!(source["result"]["structuredContent"]["text"]
.as_str()
.unwrap()
.contains("2_u64.pow"));
}
+12 -6
View File
@@ -35,6 +35,7 @@ constexpr size_t max_passage_bytes = 1024 * 1024;
constexpr size_t max_text_bytes = 65536;
constexpr size_t max_windows = 256;
constexpr int context_tokens = 2048;
constexpr int query_context_tokens = 512;
constexpr int dimensions = 768;
static void require(bool condition, const char * message) {
@@ -53,6 +54,8 @@ class Encoder {
const llama_vocab * vocab = nullptr;
mtmd_context * media = nullptr;
std::string model_path;
bool query_only;
int allocated_context;
void release() {
if (media) mtmd_free(media);
if (batch) llama_batch_ext_free(batch);
@@ -69,7 +72,7 @@ class Encoder {
model = llama_model_load_from_file(model_path.c_str(),mp);
if (!model) return false;
auto cp = llama_context_default_params();
cp.n_ctx = context_tokens; cp.n_batch = context_tokens; cp.n_ubatch = context_tokens;
cp.n_ctx = allocated_context; cp.n_batch = allocated_context; cp.n_ubatch = allocated_context;
cp.n_seq_max = 1; cp.kv_unified = true; cp.embeddings = true;
cp.offload_kqv = gpu; cp.op_offload = gpu;
cp.pooling_type = LLAMA_POOLING_TYPE_MEAN;
@@ -88,7 +91,7 @@ public:
std::string cpu_backend_module;
bool vulkan_module_loaded = false;
int threads;
Encoder(const std::string & path, int cpu_threads, bool want_gpu) : model_path(path),threads(cpu_threads) {
Encoder(const std::string & path, int cpu_threads, bool want_gpu, bool query_role) : model_path(path),query_only(query_role),allocated_context(query_role ? query_context_tokens : context_tokens),threads(cpu_threads) {
llama_log_set([](ggml_log_level level, const char * text, void *) {
if (level >= GGML_LOG_LEVEL_WARN) std::fputs(text,stderr);
},nullptr);
@@ -331,6 +334,7 @@ public:
}
json handle(const json & request) {
auto op = request.at("op").get<std::string>();
require(!query_only || op == "query" || op == "code_query", "Query worker only supports query and code_query operations");
if(op=="image" || op=="audio") {
auto encoded=request.at("data").get<std::string>();
require(encoded.size()<=12000000 && encoded.size()%4==0,"Invalid media payload length");
@@ -363,7 +367,7 @@ public:
bool debug = request.value("debug_performance",false);
json timings;
auto prefix = op == "code_query" ? "task: code retrieval | query: " : "task: search result | query: ";
auto vector = embed(prefix+request.at("text").get<std::string>(),512,debug ? &timings : nullptr);
auto vector = embed(prefix+request.at("text").get<std::string>(),query_context_tokens,debug ? &timings : nullptr);
json response = {{"vectors",json::array({vector})}};
if (debug) response["performance"] = timings;
return response;
@@ -382,17 +386,19 @@ public:
int main(int argc, char ** argv) {
try {
std::string path, device = "auto"; int threads = 1;
std::string path, device = "auto", role = "indexing"; int threads = 1;
for (int i = 1; i < argc; ++i) {
require(i+1 < argc,"Worker options require values");
std::string option = argv[i++], value = argv[i];
if (option == "--model") path = value;
else if (option == "--threads") threads = std::stoi(value);
else if (option == "--device") device = value;
else if (option == "--role") role = value;
else throw std::runtime_error("Unknown worker option");
}
require(!path.empty() && threads >= 1 && threads <= 8,"Model path required; threads must be 1–8");
require(device == "cpu" || device == "auto","Device must be cpu or auto");
require(role == "query" || role == "indexing", "Role must be query or indexing");
bool force_cpu = std::getenv("SEARCHI_FORCE_CPU") && std::string(std::getenv("SEARCHI_FORCE_CPU")) == "1";
// Disable backend registration before llama initializes its registry.
// A GPU-capable distribution can therefore be tested on CPU without
@@ -406,9 +412,9 @@ int main(int argc, char ** argv) {
require(setenv("GGML_DISABLE_VULKAN", "1", 1) == 0, "Cannot disable Vulkan for CPU startup");
#endif
}
Encoder encoder(path,threads,device == "auto" && !force_cpu);
Encoder encoder(path,threads,device == "auto" && !force_cpu,role == "query");
std::cout << json({{"ready",true},{"model","google/embeddinggemma-2"},{"dimensions",dimensions},
{"device",encoder.device},{"threads",threads},{"runtime",SEARCHI_LLAMA_TAG},{"media_supported",true},{"code_query_supported",true},
{"device",encoder.device},{"threads",threads},{"runtime",SEARCHI_LLAMA_TAG},{"media_supported",true},{"code_query_supported",true},{"role",role},{"context_tokens",role == "query" ? query_context_tokens : context_tokens},
{"cpu_backend_module",encoder.cpu_backend_module},{"vulkan_module_loaded",encoder.vulkan_module_loaded}}).dump() << '\n' << std::flush;
for (;;) {
std::string line; char c;
+108 -6
View File
@@ -29,6 +29,7 @@ pub struct ImageEmbedding {
const MODEL_FILE: &str = "embeddinggemma-2-Q8_0.gguf";
const MODEL_BYTES: u64 = 309855456;
const MODEL_SHA256: &str = "2188ac1deca4b77dffefd603c2776a9d76d9d74ec01841392982ebb840b09135";
const QUERY_CONTEXT_TOKENS: usize = 512;
const MAX_REQUEST: usize = 16 * 1024 * 1024;
const MAX_RESPONSE: usize = 4 * 1024 * 1024;
@@ -50,6 +51,14 @@ pub struct WorkerInfo {
pub media_supported: bool,
#[serde(default)]
pub code_query_supported: bool,
#[serde(default)]
pub role: String,
#[serde(default)]
pub context_tokens: usize,
}
fn supports_query_role(info: &WorkerInfo) -> bool {
info.role == "query" && info.context_tokens == QUERY_CONTEXT_TOKENS
}
fn query_operation(info: &WorkerInfo, code: bool) -> Result<&'static str> {
@@ -68,6 +77,7 @@ pub struct GemmaEmbedder {
pub info: WorkerInfo,
healthy: bool,
force_cpu: bool,
query_only: bool,
model_dir: PathBuf,
worker_path: PathBuf,
projector_verified: bool,
@@ -138,11 +148,40 @@ impl GemmaEmbedder {
Self::load_with_device(model_dir, worker, threads, Device::Auto)
}
/// A foreground-only session with a 512-token native context. The same
/// hardware-detecting worker still provides 2048-token indexing sessions.
pub fn load_query(
model_dir: impl AsRef<Path>,
worker: impl AsRef<Path>,
threads: usize,
) -> Result<Self> {
Self::load_query_with_device(model_dir, worker, threads, Device::Auto)
}
pub fn load_query_with_device(
model_dir: impl AsRef<Path>,
worker: impl AsRef<Path>,
threads: usize,
device: Device,
) -> Result<Self> {
Self::load_role(model_dir, worker, threads, device, true)
}
pub fn load_with_device(
model_dir: impl AsRef<Path>,
worker: impl AsRef<Path>,
threads: usize,
device: Device,
) -> Result<Self> {
Self::load_role(model_dir, worker, threads, device, false)
}
fn load_role(
model_dir: impl AsRef<Path>,
worker: impl AsRef<Path>,
threads: usize,
device: Device,
query_only: bool,
) -> Result<Self> {
ensure!(
(1..=8).contains(&threads),
@@ -162,13 +201,26 @@ impl GemmaEmbedder {
let force_cpu = std::env::var("SEARCHI_FORCE_CPU").is_ok_and(|value| value == "1");
let cpu = force_cpu || matches!(device, Device::Cpu);
start_with_cpu_fallback(cpu, |cpu| {
Self::start_verified(&model_dir, &worker, threads, cpu)
Self::start_verified(&model_dir, &worker, threads, cpu, query_only)
})
.with_context(|| {
if query_only {
"Cannot initialize foreground query role; repair or rebuild the native runtime"
} else {
"Cannot initialize indexing runtime"
}
})
}
// Both attempts share the already validated model/worker paths. CPU fallback
// does not repeat the expensive hash check or weaken the handshake contract.
fn start_verified(model_dir: &Path, worker: &Path, threads: usize, cpu: bool) -> Result<Self> {
fn start_verified(
model_dir: &Path,
worker: &Path,
threads: usize,
cpu: bool,
query_only: bool,
) -> Result<Self> {
let model = model_dir.join(MODEL_FILE);
let mut command = Command::new(worker);
command
@@ -183,6 +235,9 @@ impl GemmaEmbedder {
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::inherit());
if query_only {
command.args(["--role", "query"]);
}
// Background helpers must not create a visible console on Windows.
#[cfg(windows)]
{
@@ -235,7 +290,8 @@ impl GemmaEmbedder {
return Err(error).context("Invalid Gemma worker handshake");
}
};
if !info.ready
if (query_only && (!supports_query_role(&info)))
|| !info.ready
|| info.model != "google/embeddinggemma-2"
|| info.dimensions != GEMMA_DIMENSIONS
|| info.runtime != "b11461-no-logits-v1"
@@ -246,7 +302,7 @@ impl GemmaEmbedder {
let _ = child.kill();
let _ = child.wait();
let _ = reader_thread.join();
anyhow::bail!("Gemma worker handshake violates configured model/device contract");
anyhow::bail!("Gemma worker handshake violates configured model/device contract or requested role; repair or rebuild the native runtime");
}
let (requests, writes, writer_thread) = spawn_writer(stdin);
Ok(Self {
@@ -259,6 +315,7 @@ impl GemmaEmbedder {
info,
healthy: true,
force_cpu: cpu,
query_only,
model_dir: model_dir.to_path_buf(),
worker_path: worker.to_path_buf(),
projector_verified: false,
@@ -288,11 +345,12 @@ impl GemmaEmbedder {
// Restart once as CPU and retry this same bounded request.
let _ = self.child.wait();
let restart_started = self.query_profile.as_ref().map(|_| Instant::now());
let replacement = Self::load_with_device(
let replacement = Self::load_role(
&self.model_dir,
&self.worker_path,
self.info.threads,
Device::Cpu,
self.query_only,
)
.with_context(|| {
format!("CPU worker restart failed after GPU transport error: {error}")
@@ -406,6 +464,10 @@ impl GemmaEmbedder {
}
pub fn split_windows(&mut self, text: &str, title: &str) -> Result<Vec<TextWindow>> {
ensure!(
!self.query_only,
"Foreground query worker cannot index documents or media; use an indexing session"
);
ensure!(
!text.trim().is_empty() && text.len() <= 1024 * 1024,
"Passage must be nonempty and at most 1 MiB"
@@ -432,6 +494,10 @@ impl GemmaEmbedder {
}
pub fn embed_documents(&mut self, texts: &[&str], title: &str) -> Result<Vec<Vec<f32>>> {
ensure!(
!self.query_only,
"Foreground query worker cannot index documents or media; use an indexing session"
);
ensure!(
!texts.is_empty() && texts.len() <= 8,
"Document batch size must be between 1 and 8"
@@ -545,6 +611,10 @@ impl GemmaEmbedder {
paths: &[&Path],
budget: u32,
) -> Result<Vec<ImageEmbedding>> {
ensure!(
!self.query_only,
"Foreground query worker cannot index documents or media; use an indexing session"
);
ensure!(
!paths.is_empty() && paths.len() <= 8,
"Image batch must contain 1–8 paths"
@@ -567,6 +637,10 @@ impl GemmaEmbedder {
/// One mono 16 kHz normalized PCM window, at most 30 seconds. Callers
/// segment entire recordings and preserve real timestamps; no truncation.
pub fn embed_audio_pcm(&mut self, samples: &[f32]) -> Result<Vec<f32>> {
ensure!(
!self.query_only,
"Foreground query worker cannot index documents or media; use an indexing session"
);
ensure!(
!samples.is_empty() && samples.len() <= 480000,
"Audio window must contain 1–480000 mono 16 kHz samples"
@@ -851,9 +925,12 @@ mod tests {
runtime: "fixture".into(),
media_supported: false,
code_query_supported: false,
role: "indexing".into(),
context_tokens: 2048,
},
healthy: true,
force_cpu: true,
query_only: false,
model_dir: PathBuf::new(),
worker_path: PathBuf::new(),
projector_verified: false,
@@ -861,6 +938,31 @@ mod tests {
}
}
#[test]
fn query_role_capability_is_explicit_and_independent_of_media_support() {
let mut info: WorkerInfo = serde_json::from_value(json!({
"ready":true,"model":"google/embeddinggemma-2","dimensions":768,
"device":"cpu","threads":1,"runtime":"b11461-no-logits-v1",
"media_supported":true,"code_query_supported":true
}))
.unwrap();
assert!(
!supports_query_role(&info),
"Legacy worker cannot claim reduced query allocation"
);
info.role = "query".into();
info.context_tokens = 2048;
assert!(!supports_query_role(&info));
info.context_tokens = 512;
assert!(supports_query_role(&info));
assert!(
info.media_supported,
"Binary capability remains available to full sessions"
);
info.role = "indexing".into();
assert!(!supports_query_role(&info));
}
#[test]
fn disconnected_writer_queue_is_transport_failure_and_reaps_helper() {
let mut fixture = stalled_protocol_helper(false);
@@ -962,7 +1064,7 @@ mod tests {
let info: WorkerInfo = serde_json::from_value(handshake)?;
ensure!(
info.model == "google/embeddinggemma-2",
"Gemma worker handshake violates configured model/device contract"
"Gemma worker handshake violates configured model/device contract or requested role; repair or rebuild the native runtime"
);
Ok(())
});
@@ -0,0 +1,119 @@
use anyhow::Result;
use searchi_semantic::{Device, GemmaEmbedder};
fn cosine(a: &[f32], b: &[f32]) -> f64 {
a.iter()
.zip(b)
.map(|(a, b)| f64::from(*a) * f64::from(*b))
.sum()
}
#[test]
#[ignore = "Requires verified Google model and current CPU-capable native worker"]
fn query_context_preserves_real_vectors_ranking_and_token_boundary() -> Result<()> {
let model = std::env::var_os("SEARCHI_GEMMA_MODEL_DIR").expect("Set SEARCHI_GEMMA_MODEL_DIR");
let worker = std::env::var_os("SEARCHI_GEMMA_WORKER").expect("Set SEARCHI_GEMMA_WORKER");
let mut full = GemmaEmbedder::load_with_device(&model, &worker, 1, Device::Cpu)?;
let mut query = GemmaEmbedder::load_query_with_device(&model, &worker, 1, Device::Cpu)?;
assert_eq!(full.info.context_tokens, 2048);
assert_eq!(query.info.context_tokens, 512);
assert_eq!(query.info.role, "query");
// This describes the binary's media capability, not this session's role.
assert!(query.info.media_supported);
let documents = full.embed_documents(
&[
"The supplier agreement may be terminated with sixty days written notice.",
"Business train tickets and hotel stays are reimbursed after submitting receipts.",
"Die Gartenplanung sieht ein Hochbeet und eine Regentonne vor.",
],
"none",
)?;
for (text, expected) in [
("Wie beenden wir die Zusammenarbeit mit dem Lieferanten?", 0),
("How do I claim my accommodation and railway expenses?", 1),
(
"Where is the plan for the vegetable bed and rainwater barrel?",
2,
),
] {
let before = full.embed_query(text)?;
let after = query.embed_query(text)?;
let parity = cosine(&before, &after);
assert!(parity >= 0.99999, "Context changed vector: {parity}");
let rank = |vector: &[f32]| {
documents
.iter()
.enumerate()
.max_by(|a, b| cosine(a.1, vector).total_cmp(&cosine(b.1, vector)))
.unwrap()
.0
};
assert_eq!(rank(&before), expected);
assert_eq!(rank(&after), expected);
println!("Query role parity {parity:.8}; expected document {expected}");
}
for text in [
"retry failed requests with exponential backoff",
"Zugriffstoken bei Ablauf erneuern",
"متى تغلق المكتبة",
"ইলিশ মাছ রান্না",
] {
assert!(
cosine(
&full.embed_code_query(text)?,
&query.embed_code_query(text)?
) >= 0.99999
);
}
// Actual token counts determine the exact boundary, rather than assuming
// words equal tokens. Each repeated plain token contributes one token.
let mut repetitions = 490usize;
let mut boundary = String::new();
for _ in 0..8 {
let text = " cat".repeat(repetitions);
full.embed_query_debug(&text)?;
let tokens = full.query_diagnostics()["worker_token_count"] as usize;
if tokens == 512 {
boundary = text;
break;
}
assert!(tokens < 512, "Unexpected fixture token count {tokens}");
repetitions += 512 - tokens;
}
assert!(
!boundary.is_empty(),
"Could not construct exact512token query"
);
let before = full.embed_query(&boundary)?;
let after = query.embed_query_debug(&boundary)?;
assert_eq!(query.query_diagnostics()["worker_token_count"], 512);
assert!(cosine(&before, &after) >= 0.99999);
assert!(query.embed_query(&(boundary + " cat")).is_err());
assert!(query
.embed_query("supplier cancellation")?
.iter()
.all(|v| v.is_finite()));
assert!(query
.embed_documents(&["document"], "none")
.unwrap_err()
.to_string()
.contains("Foreground query worker"));
assert!(query.split_windows("document", "none").is_err());
assert!(query
.embed_audio_pcm(&[0.0; 16])
.unwrap_err()
.to_string()
.contains("Foreground query worker"));
assert!(query
.embed_images(&[std::path::Path::new("missing.png")])
.unwrap_err()
.to_string()
.contains("Foreground query worker"));
// Full2048-token indexing survives, including long document extraction.
assert!(
full.split_windows(&"ordinary maintenance log ".repeat(1000), "log")?
.len()
> 1
);
Ok(())
}
+11
View File
@@ -314,6 +314,17 @@ impl HybridSearch {
Self::with_provider(db, Box::new(embedder))
}
/// Foreground-only Google session. Indexing must use `open` instead.
pub fn open_query(
db: impl AsRef<Path>,
model_dir: impl AsRef<Path>,
runtime_library: impl AsRef<Path>,
threads: usize,
) -> Result<Self> {
let embedder = GemmaEmbedder::load_query(model_dir, runtime_library, threads)?;
Self::with_provider(db, Box::new(embedder))
}
/// Explicit provider injection for future model generations, never fallback.
pub fn with_provider(
db: impl AsRef<Path>,