From 393a0aff62a450e973112be05bb73b4b443561d9 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 31 Oct 2026 06:40:22 +0100 Subject: [PATCH] live smoke: refresh one small public repo into a scratch source collection, search it, delete it; the Qdrant source port can be pointed at a scratch collection Co-Authored-By: Claude Sonnet 5.5 --- crates/huginn-daemon/src/index/qdrant.rs | 29 +++++++--- .../huginn-daemon/tests/live_index_smoke.rs | 57 +++++++++++++++++++ 2 files changed, 79 insertions(+), 7 deletions(-) diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index 5dc45eb..b7c1d9a 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -15,6 +15,9 @@ use super::{CollectionMeta, INDEX_CALL_TIMEOUT, Described, Hit, Http, Method, Po /// never lists or touches another. pub struct QdrantIndex { http: Http, + /// The collection the source port reads and writes: `huginn_sources`, + /// unless a live test points it at a scratch one. + sources: String, } /// The ids fetched per scroll page. @@ -22,7 +25,13 @@ const SCROLL_PAGE: u64 = 1000; impl QdrantIndex { pub fn new(base_url: &str) -> Self { - Self { http: Http::new(base_url, INDEX_CALL_TIMEOUT) } + Self { http: Http::new(base_url, INDEX_CALL_TIMEOUT), sources: COLLECTION.to_owned() } + } + + /// The source port on another collection, for a live test's scratch one. + pub fn with_sources_collection(mut self, collection: &str) -> Self { + self.sources = collection.to_owned(); + self } fn ask(&self, method: Method, path: &str, body: Option<&Value>, ok: &[u16]) -> Result<(u16, Value)> { @@ -222,8 +231,9 @@ fn sources_described(metadata: &Value) -> DescribedSources { impl SourceIndex for QdrantIndex { fn describe_sources(&mut self, within: Duration) -> Result { + let collection = self.sources.clone(); let http = self.http.bounded(within); - let (status, answer) = self.ask_with(&http, Method::Get, &format!("/collections/{COLLECTION}"), None, &[200, 404])?; + let (status, answer) = self.ask_with(&http, Method::Get, &format!("/collections/{collection}"), None, &[200, 404])?; if status == 404 { return Ok(DescribedSources::Absent); } @@ -231,7 +241,8 @@ impl SourceIndex for QdrantIndex { } fn recreate_sources(&mut self, meta: &SourcesMeta) -> Result<()> { - let path = format!("/collections/{COLLECTION}"); + let collection = self.sources.clone(); + let path = format!("/collections/{collection}"); self.ask(Method::Delete, &format!("{path}?wait=true"), None, &[200, 404])?; let body = json!({ "vectors": { "size": meta.dimensions, "distance": "Cosine", "on_disk": true }, @@ -254,40 +265,44 @@ impl SourceIndex for QdrantIndex { } fn set_cursors(&mut self, cursors: &BTreeMap) -> Result<()> { + let collection = self.sources.clone(); // Qdrant merges metadata by top-level key and replaces a nested object // whole, so this is the whole cursors object and nothing beside it. let body = json!({ "metadata": { "cursors": cursors } }); - self.ask(Method::Patch, &format!("/collections/{COLLECTION}?wait=true"), Some(&body), &[200])?; + self.ask(Method::Patch, &format!("/collections/{collection}?wait=true"), Some(&body), &[200])?; Ok(()) } fn delete_by(&mut self, repo: &str, paths: Option<&[String]>) -> Result<()> { + let collection = self.sources.clone(); let mut must = vec![json!({ "key": "repo", "match": { "value": repo } })]; if let Some(paths) = paths { must.push(json!({ "key": "path", "match": { "any": paths } })); } let body = json!({ "filter": { "must": must } }); - self.ask(Method::Post, &format!("/collections/{COLLECTION}/points/delete?wait=true"), Some(&body), &[200])?; + self.ask(Method::Post, &format!("/collections/{collection}/points/delete?wait=true"), Some(&body), &[200])?; Ok(()) } fn upsert_sources(&mut self, points: &[SourcePoint]) -> Result<()> { + let collection = self.sources.clone(); let points: Vec = points .iter() .map(|point| json!({ "id": point.id, "vector": point.vector, "payload": payload_json(&point.payload) })) .collect(); - self.ask(Method::Put, &format!("/collections/{COLLECTION}/points?wait=true"), Some(&json!({ "points": points })), &[200])?; + self.ask(Method::Put, &format!("/collections/{collection}/points?wait=true"), Some(&json!({ "points": points })), &[200])?; Ok(()) } fn search_sources(&mut self, vector: &[f32], repo: Option<&str>, limit: u32, within: Duration) -> Result> { + let collection = self.sources.clone(); let http = self.http.bounded(within); let mut body = json!({ "vector": vector, "limit": limit, "with_payload": true, "with_vector": false }); if let Some(repo) = repo { body["filter"] = json!({ "must": [{ "key": "repo", "match": { "value": repo } }] }); } let (_, answer) = - self.ask_with(&http, Method::Post, &format!("/collections/{COLLECTION}/points/search"), Some(&body), &[200])?; + self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?; let found = answer["result"].as_array().context("Qdrant's search has no result")?; found .iter() diff --git a/crates/huginn-daemon/tests/live_index_smoke.rs b/crates/huginn-daemon/tests/live_index_smoke.rs index c6a2c9e..8c2b7e6 100644 --- a/crates/huginn-daemon/tests/live_index_smoke.rs +++ b/crates/huginn-daemon/tests/live_index_smoke.rs @@ -132,3 +132,60 @@ fn serde_json_free_query(vector: &[f32]) -> String { let numbers: Vec = vector.iter().map(|value| value.to_string()).collect(); format!("{{\"vector\":[{}],\"limit\":1}}", numbers.join(",")) } + +/// One forge repo, however many the forge lists: the live check refreshes a +/// single small repo, not the corpus. +struct OneRepo { + inner: huginn_daemon::sources::forge::HttpForge, + name: String, +} + +impl huginn_daemon::sources::Forge for OneRepo { + fn catalog(&mut self) -> anyhow::Result> { + Ok(self.inner.catalog()?.into_iter().filter(|repo| repo.name == self.name).collect()) + } + + fn clone_url(&self, name: &str) -> String { + self.inner.clone_url(name) + } +} + +/// Refreshes one small public repo into a scratch source collection, searches +/// it, and deletes the collection. Needs `HUGINN_SMOKE_FORGE` (the forge's +/// loopback address, so run it on Yggdrasil) and `HUGINN_SMOKE_REPO` beside +/// the other three variables. The collection is the scratch one alone; +/// `huginn_sources` is never named. +#[test] +#[ignore = "needs a live Qdrant, Ollama and forge, and writes a scratch collection"] +fn a_small_repo_is_refreshed_into_a_scratch_source_collection_and_found() { + use huginn_daemon::sources::{Forge, forge::HttpForge, refresh}; + let (qdrant, ollama, model) = (env("HUGINN_SMOKE_QDRANT"), env("HUGINN_SMOKE_OLLAMA"), env("HUGINN_SMOKE_MODEL")); + let (forge, repo) = (env("HUGINN_SMOKE_FORGE"), env("HUGINN_SMOKE_REPO")); + let stamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos(); + let scratch = format!("huginn-smoke-sources-{:x}-{stamp:x}", std::process::id()); + let root = std::env::temp_dir().join(&scratch); + + let mut embedder = OllamaEmbedder::new(&ollama, &model); + let mut index = QdrantIndex::new(&qdrant).with_sources_collection(&scratch); + let mut forge = OneRepo { inner: HttpForge::new(&forge), name: repo.clone() }; + assert_eq!(forge.catalog().unwrap().len(), 1, "{repo} is a public, live, non-empty repo on this forge"); + let mut lines = Vec::new(); + let refreshed = refresh(&mut embedder, &mut index, &mut forge, &root, &mut |report| lines.push(report.to_string())); + let found = refreshed + .as_ref() + .ok() + .and_then(|_| { + let query = huginn_mind::SourceQuery { + text: huginn_mind::eureka_pipeline::Line("what does this repository do".into()), + top_k: 3, + repo: None, + }; + huginn_daemon::sources::search(&mut embedder, &mut index, &query, Duration::from_secs(60)).ok() + }); + let _ = raw("DELETE", &format!("{qdrant}/collections/{scratch}?wait=true"), None); + let _ = std::fs::remove_dir_all(&root); + println!("journal: {lines:?}"); + assert!(refreshed.unwrap().failed.is_empty()); + let found = found.expect("a search answers from the scratch collection"); + assert!(!found.hits.is_empty() && found.hits.iter().all(|hit| hit.repo == repo)); +}