diff --git a/crates/eureka-state/src/main.rs b/crates/eureka-state/src/main.rs index dcf65c6..7f9c7b7 100644 --- a/crates/eureka-state/src/main.rs +++ b/crates/eureka-state/src/main.rs @@ -281,7 +281,7 @@ impl EurekaState { } #[tool( - description = "Query the mind with a CultNet typed selection, optionally ranked by semantic nearness. Returns one page; pass its cursor back in the selection for the next." + description = "Query the mind with a CultNet typed selection, optionally ranked by semantic nearness. Returns one page; pass its cursor back in the selection for the next. A semantic answer returns at most top_k members of the selection, nearest first, with scores (one cosine similarity per row, in order)." )] async fn query(&self, Parameters(input): Parameters) -> CallToolResult { self.answer(move |instance| HuginnMindRequest::Query { diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index 6dc606e..f2feb80 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -165,7 +165,7 @@ fn ruling(quote: Option<&str>) -> PipelineDocument { } fn empty_page() -> PipelineSelectionPage { - PipelineSelectionPage { matched: 0, as_of: 1, next: None, items: PipelinePageItems::Headers(vec![]), edges: None } + PipelineSelectionPage { matched: 0, as_of: 1, next: None, items: PipelinePageItems::Headers(vec![]), edges: None, scores: None } } fn answer(operation: &str, response: &HuginnMindResponse) -> cultnet_rs::CultNetMessage { @@ -524,6 +524,7 @@ fn committing(stored: Vec) -> Server { next: None, items: PipelinePageItems::Documents(stored.clone()), edges: None, + scores: None, }; answer("query", &HuginnMindResponse::Query(page)) } diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs index cb09085..b2e2be8 100644 --- a/crates/eureka-state/tests/common/mod.rs +++ b/crates/eureka-state/tests/common/mod.rs @@ -89,7 +89,7 @@ impl IndexSink for Inert { fn status(&self) -> IndexStatus { IndexStatus::Current } - fn search(&mut self, _text: &str, _top_k: u32) -> Result { + fn search(&mut self, _text: &str, _top_k: u32, _among: Vec) -> Result { Err("this index cannot search".into()) } fn search_deadline(&self) -> Duration { diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index afe0185..d3a1861 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -6,7 +6,7 @@ use chrono::{DateTime, Utc}; use cultnet_rs::Selection; use huginn_mind::eureka_pipeline::{PipelineRef, Slug}; use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, IndexStatus}; -use huginn_mind::{Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery}; +use huginn_mind::{Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery}; /// Where the index attaches. Called after a batch commits, with the refs the /// batch landed, derived writes included; the sink may read each through @@ -22,9 +22,11 @@ use huginn_mind::{Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, Semant pub trait IndexSink { fn committed(&mut self, mind: &Mind, writes: &[PipelineRef]) -> anyhow::Result<()>; fn status(&self) -> IndexStatus; - /// Starts a search for the documents nearest `text`, at most `top_k` of - /// them wanted. `Err` says why the index cannot be asked now. - fn search(&mut self, text: &str, top_k: u32) -> Result; + /// Starts a search for the documents nearest `text`: the index scores only + /// `among`, the member document ids the mind chose, and answers the + /// nearest first, at most `top_k`. `Err` says why the index cannot be + /// asked now. + fn search(&mut self, text: &str, top_k: u32, among: Vec) -> Result; /// How long after it was asked a search is worth answering: the index's /// own bound on it, which the loop waiting for the answer must share. fn search_deadline(&self) -> std::time::Duration; @@ -53,6 +55,7 @@ pub struct Search { pub ticket: SearchTicket, selection: Selection, semantic: SemanticQuery, + members: Members, } /// What a request came to: an answer, or a search the index has yet to @@ -171,15 +174,26 @@ impl> Daemon { Handled::Answered(answer) } - /// A semantic query starts only if the mind would accept it (so nothing is - /// embedded for a query that cannot be ranked) and the index can be asked. + /// A semantic query starts only if the mind accepts it, and the mind + /// decides its members before the index is asked anything. A selection + /// with no members is answered at once: nothing is embedded and the index + /// is not called. Otherwise the index scores those members and the answer + /// waits in a `Search`. fn begin_search(&mut self, selection: Selection, semantic: SemanticQuery) -> Handled { let refused = |refusal: MindRefusal| Handled::Answered(HuginnMindResponse::Refused(refusal)); - if let Err(refusal) = self.mind.check_semantic(&selection, &semantic) { - return refused(refusal); + let members = match self.mind.members(&selection, &semantic) { + Ok(members) => members, + Err(refusal) => return refused(refusal), + }; + if members.ids.is_empty() { + return match self.mind.rank(&selection, &semantic, &members, &[]) { + Ok(ranked) => Handled::Answered(HuginnMindResponse::Query(ranked.page)), + Err(refusal) => refused(refusal), + }; } - match self.index.search(&semantic.text.0, semantic.top_k) { - Ok(ticket) => Handled::Searching(Search { ticket, selection, semantic }), + let among = members.ids.iter().map(|id| id.id.0.clone()).collect(); + match self.index.search(&semantic.text.0, semantic.top_k, among) { + Ok(ticket) => Handled::Searching(Search { ticket, selection, semantic, members }), Err(detail) => refused(MindRefusal::Unavailable { detail }), } } @@ -199,16 +213,16 @@ impl> Daemon { self.index.abandon(ticket); } - /// The answer to a search: the index's candidates ranked through the mind - /// (`Mind::rank` drops what the mind does not hold and lets the selection - /// decide the rest), or, when the index could not answer, a typed + /// The answer to a search: the index's scores for the members ranked + /// through the mind (`Mind::rank` answers as of the snapshot the members + /// were taken at and drops what is not a member), or, when the index could not answer, a typed /// `Unavailable`. Never an empty page in place of a failure. pub fn finish(&mut self, search: Search, found: Result) -> HuginnMindResponse { let hits = match found { Ok(hits) => hits, Err(detail) => return HuginnMindResponse::Refused(MindRefusal::Unavailable { detail }), }; - match self.mind.rank(&search.selection, &search.semantic, &hits) { + match self.mind.rank(&search.selection, &search.semantic, &search.members, &hits) { Ok(ranked) => { if let Some(note) = dropped_note(&ranked.dropped) { eprintln!("{note}"); @@ -479,7 +493,7 @@ pub(crate) mod tests { IndexStatus::Current } - fn search(&mut self, _text: &str, _top_k: u32) -> Result { + fn search(&mut self, _text: &str, _top_k: u32, _among: Vec) -> Result { Err("this index cannot search".into()) } @@ -521,7 +535,7 @@ pub(crate) mod tests { IndexStatus::Current } - fn search(&mut self, _text: &str, _top_k: u32) -> Result { + fn search(&mut self, _text: &str, _top_k: u32, _among: Vec) -> Result { Err("this index cannot search".into()) } @@ -547,7 +561,7 @@ pub(crate) mod tests { IndexStatus::Current } - fn search(&mut self, _text: &str, _top_k: u32) -> Result { + fn search(&mut self, _text: &str, _top_k: u32, _among: Vec) -> Result { Err("this index cannot search".into()) } diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index 20bcc8c..dd6fc09 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -8,7 +8,7 @@ use std::time::Duration; use anyhow::{Result, bail, ensure}; -use super::{CollectionMeta, Described, Embedder, Hit, ModelIdentity, Point, VectorIndex}; +use super::{CollectionMeta, Described, Embedder, Hit, ModelIdentity, Point, VectorIndex, point_id}; pub(crate) fn identity(digest: &str) -> ModelIdentity { ModelIdentity { name: "model".into(), digest: digest.into(), dimensions: 4 } @@ -147,6 +147,8 @@ pub(crate) struct IndexState { pub(crate) hits: Vec, /// The collection, vector length and limit of every search, in order. pub(crate) searches: Vec<(String, usize, u32)>, + /// The point ids each search was restricted to, in order. + pub(crate) among: Vec>, /// How many times every point id was listed: only a reconciliation does. pub(crate) listed: u32, /// Whether a search fails at the vector store, the rest of it working. @@ -218,18 +220,20 @@ impl VectorIndex for FakeIndex { Ok(()) } - fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result> { + fn search(&mut self, collection: &str, vector: &[f32], among: &[String], limit: u32, within: Duration) -> Result> { let mut state = self.state.lock().unwrap(); state.bounds.push(("search", within)); if state.down { bail!("the vector store is down"); } state.searches.push((collection.into(), vector.len(), limit)); + state.among.push(among.to_vec()); ensure!(!state.fail_search, "the vector store failed the search"); ensure!(state.collections.contains_key(collection), "no collection {collection}"); let held = state.collections[collection].points.len(); state.held_when_searched.push(held); - Ok(state.hits.clone()) + // A filtered exact index: only the points handed in are scored. + Ok(state.hits.iter().filter(|hit| among.contains(&point_id(&hit.doc_id))).take(limit as usize).cloned().collect()) } } diff --git a/crates/huginn-daemon/src/index/mod.rs b/crates/huginn-daemon/src/index/mod.rs index d52f5b6..7f12f57 100644 --- a/crates/huginn-daemon/src/index/mod.rs +++ b/crates/huginn-daemon/src/index/mod.rs @@ -129,9 +129,6 @@ pub const BULK_EMBED_TIMEOUT: Duration = Duration::from_secs(300); /// adapter's default. pub const INDEX_CALL_TIMEOUT: Duration = Duration::from_secs(60); -/// The most candidates one search asks the index for. -pub const OVERSAMPLE_MAX: u32 = 200; - /// What an embedder is: the model by name and by digest, and the length of the /// vectors it returns. Compatibility of a collection is decided on this and /// on nothing else: never on the address the embedder was reached at. @@ -208,10 +205,11 @@ pub trait VectorIndex { /// Every point id the collection holds. fn ids(&mut self, collection: &str) -> Result>; fn upsert(&mut self, collection: &str, points: &[Point]) -> Result<()>; - /// The `limit` nearest points to `vector`, best first, with no filter: which - /// documents may be shown is the mind's to decide. An error once `within` - /// has passed. - fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result>; + /// Scores exactly the points in `among` (exact, not approximate) against + /// `vector` and returns the `limit` nearest, best first. Which documents + /// are members is the mind's: the index only scores the points it is + /// handed. An error once `within` has passed. + fn search(&mut self, collection: &str, vector: &[f32], among: &[String], limit: u32, within: Duration) -> Result>; } /// The mind's collection: one per instance. @@ -556,11 +554,12 @@ impl Projector { Ok(Ok(vectors)) } - /// The candidates a query text finds: the text is embedded with the model's - /// query instruction and the collection is searched, with no filter and an - /// oversample of `min(top_k * 4, 200)`, because the mind will drop most of - /// what the index cannot know is not in force. A hit naming a kind this - /// organ has no name for is logged and left out. + /// The candidates a query text finds: the text is embedded with the model + /// query instruction and the collection scores exactly the `among` member + /// documents (ids the mind chose) and returns the `top_k` nearest. There + /// is no oversample: the members are scored, so the nearest `top_k` of them + /// are the answer. A hit naming a kind this organ has no name for is + /// logged and left out. /// /// A search is refused while the index is being rebuilt or cannot be asked /// (`unsearchable`), naming the state and how many documents are pending. @@ -573,7 +572,7 @@ impl Projector { /// was reconciled under. Every call is bounded by what remains of /// `within`. A search that fails for any other reason (a timeout, a store /// error) fails alone and leaves the reconciliation standing. - pub fn search(&mut self, text: &str, top_k: u32, within: Duration) -> Result { + pub fn search(&mut self, text: &str, top_k: u32, among: &[String], within: Duration) -> Result { if let Some(why) = unsearchable(&self.progress_status()) { bail!("{why}"); } @@ -592,8 +591,8 @@ impl Projector { ensure!(vectors.len() == 1, "the embedder returned {} vectors for one query", vectors.len()); let vector = vectors.remove(0); ensure!(vector.len() == dimensions, "the embedder returned a query vector of {} where the collection holds {dimensions}", vector.len()); - let limit = top_k.saturating_mul(4).min(OVERSAMPLE_MAX); - let found = self.index.search(&self.collection, &vector, limit, budget.left()?)?; + let points: Vec = among.iter().map(|id| point_id(id)).collect(); + let found = self.index.search(&self.collection, &vector, &points, top_k, budget.left()?)?; let mut hits = Vec::with_capacity(found.len()); for hit in found { match PipelineKind::ALL.iter().find(|kind| kind.name() == hit.kind) { @@ -665,6 +664,8 @@ struct Queued { ticket: SearchTicket, text: String, top_k: u32, + /// The member document ids the index is to score. + among: Vec, due: Instant, } @@ -780,9 +781,9 @@ fn serve_search(projector: &mut Projector, sh } #[cfg(test)] tests::after_absorb(&mut guard); - let Some(Queued { ticket, text, top_k, due }) = take_search(projector, &mut guard) else { return }; + let Some(Queued { ticket, text, top_k, among, due }) = take_search(projector, &mut guard) else { return }; drop(guard); - let found = projector.search(&text, top_k, due.saturating_duration_since(Instant::now())).map_err(|error| format!("{error:#}")); + let found = projector.search(&text, top_k, &among, due.saturating_duration_since(Instant::now())).map_err(|error| format!("{error:#}")); lock(shared).found.push((ticket, found)); } @@ -1010,7 +1011,7 @@ impl IndexSink for WorkerSink { /// queue: the refusal is the answer, naming why and how many documents are /// pending. Documents admitted and not yet written are not a refusal: the /// search waits behind them. - fn search(&mut self, text: &str, top_k: u32) -> Result { + fn search(&mut self, text: &str, top_k: u32, among: Vec) -> Result { let mut guard = lock(&self.shared); let status = including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX)); if let Some(why) = unsearchable(&status) { @@ -1024,7 +1025,7 @@ impl IndexSink for WorkerSink { } let ticket = SearchTicket(guard.next_ticket); guard.next_ticket += 1; - guard.searches.push_back(Queued { ticket, text: text.to_owned(), top_k, due: Instant::now() + self.deadline }); + guard.searches.push_back(Queued { ticket, text: text.to_owned(), top_k, among, due: Instant::now() + self.deadline }); drop(guard); self.shared.1.notify_all(); Ok(ticket) diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index f4a1c85..b0f637e 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -138,9 +138,16 @@ impl VectorIndex for QdrantIndex { Ok(()) } - fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result> { + fn search(&mut self, collection: &str, vector: &[f32], among: &[String], limit: u32, within: Duration) -> Result> { let http = self.http.bounded(within); - let body = json!({ "vector": vector, "limit": limit, "with_payload": ["doc_id", "kind"], "with_vector": false }); + let body = json!({ + "vector": vector, + "filter": { "must": [{ "has_id": among }] }, + "params": { "exact": true }, + "limit": limit, + "with_payload": ["doc_id", "kind"], + "with_vector": false, + }); let (_, answer) = 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")?; diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs index 2e53a16..2e70044 100644 --- a/crates/huginn-daemon/src/index/tests.rs +++ b/crates/huginn-daemon/src/index/tests.rs @@ -361,7 +361,7 @@ fn every_call_on_the_search_path_is_bounded_by_the_searchs_time_and_the_flush_by index.state.lock().unwrap().bounds.clear(); let within = Duration::from_secs(20); - projector.search("anything", 3, within).unwrap(); + projector.search("anything", 3, &[], within).unwrap(); let embedder = embedder.state.lock().unwrap(); let index = index.state.lock().unwrap(); assert_eq!(embedder.identity_bounds.len(), 2, "the model is read before and after the embed"); @@ -398,7 +398,7 @@ fn a_hung_embedder_holds_the_worker_no_longer_than_the_search_that_asked() { embedder.state.lock().unwrap().hang_identity = true; let started = Instant::now(); let due = started + Duration::from_millis(400); - lock(&sink.shared).searches.push_back(Queued { ticket: SearchTicket(70), text: "hung".into(), top_k: 3, due }); + lock(&sink.shared).searches.push_back(Queued { ticket: SearchTicket(70), text: "hung".into(), top_k: 3, among: vec![], due }); sink.shared.1.notify_all(); let answered = answers(&mut sink, 1); assert_eq!(answered[0].0, SearchTicket(70)); @@ -411,6 +411,7 @@ fn a_hung_embedder_holds_the_worker_no_longer_than_the_search_that_asked() { ticket: SearchTicket(71), text: "late".into(), top_k: 3, + among: vec![], due: Instant::now().checked_sub(Duration::from_secs(1)).unwrap_or_else(Instant::now), }); sink.shared.1.notify_all(); @@ -1024,10 +1025,11 @@ fn an_idle_worker_finds_a_lost_collection_and_a_changed_model_without_a_write() } /// A search embeds the query behind the model's instruction, asks the index -/// with no filter for `min(top_k * 4, 200)`, and hands back candidates by id -/// and kind; a hit of a kind this organ has no name for is left out. +/// for exactly `top_k` among the members it was handed (no oversample), and +/// hands back candidates by id and kind; a hit of a kind this organ has no +/// name for is left out. #[test] -fn a_search_embeds_the_query_with_its_instruction_and_oversamples() { +fn a_search_embeds_the_query_with_its_instruction_and_asks_for_top_k_among_the_members() { let (_root, mind) = mind_with_documents(); let entries = mind.index_entries(None).unwrap(); let (embedder, index) = pair(); @@ -1037,19 +1039,23 @@ fn a_search_embeds_the_query_with_its_instruction_and_oversamples() { Hit { doc_id: "eureka-state:mystery:X".into(), kind: "mystery".into(), score: 0.8 }, ]; - let hits = projector.search("who owns the state", 3, SEARCH_DEADLINE).unwrap(); + let mut among: Vec = entries.iter().map(|entry| entry.id.id.0.clone()).collect(); + among.push("eureka-state:mystery:X".into()); + let hits = projector.search("who owns the state", 3, &among, SEARCH_DEADLINE).unwrap(); assert_eq!(hits.len(), 1, "the unknown kind is left out"); assert_eq!(hits[0].0.id.0, "eureka-state:question:Q1"); assert_eq!(hits[0].0.kind, PipelineKind::Question); assert_eq!(hits[0].1, 0.9); let embedded = embedder.texts(); assert_eq!(embedded.last().unwrap(), &format!("Instruct: {QUERY_INSTRUCTION}\nQuery: who owns the state")); - projector.search("again", 100, SEARCH_DEADLINE).unwrap(); - let searches = index.state.lock().unwrap().searches.clone(); - assert_eq!(searches.iter().map(|(name, length, limit)| (name.as_str(), *length, *limit)).collect::>(), [ - (collection().as_str(), 4, 12), - (collection().as_str(), 4, 200), - ]); + projector.search("again", 100, &among, SEARCH_DEADLINE).unwrap(); + let state = index.state.lock().unwrap(); + assert_eq!(state.searches.iter().map(|(name, length, limit)| (name.as_str(), *length, *limit)).collect::>(), [ + (collection().as_str(), 4, 3), + (collection().as_str(), 4, 100), + ], "top_k as asked, with no oversample"); + let points: Vec = among.iter().map(|id| point_id(id)).collect(); + assert_eq!(state.among, [points.clone(), points], "the index was handed the members' point ids"); } /// A search is never run against a collection that is not known to fit the @@ -1061,31 +1067,31 @@ fn a_search_refuses_when_the_index_is_not_reconciled_with_the_model() { let entries = mind.index_entries(None).unwrap(); let (embedder, index) = pair(); let mut fresh = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into())); - assert!(format!("{:#}", fresh.search("anything", 3, SEARCH_DEADLINE).unwrap_err()).contains("not been built")); + assert!(format!("{:#}", fresh.search("anything", 3, &[], SEARCH_DEADLINE).unwrap_err()).contains("not been built")); assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty()); let mut projector = projected(&embedder, &index, &entries); embedder.state.lock().unwrap().identity.digest = "d2".into(); let before = embedder.texts().len(); - assert!(format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err()).contains("model changed")); + assert!(format!("{:#}", projector.search("anything", 3, &[], SEARCH_DEADLINE).unwrap_err()).contains("model changed")); assert_eq!(embedder.texts().len(), before, "the query was not embedded under the new model"); assert!(index.state.lock().unwrap().searches.is_empty()); assert!(projector.stale(), "and the reconciliation is to be redone"); } -/// The search asks for the nearest points to a vector with no filter of any -/// kind (which documents may be shown is the mind's), and reads back the +/// The search scores exactly the points it is handed (a has_id filter, exact +/// search; which documents are members is the mind's), and reads back the /// document id, the kind and the score, in the order Qdrant gave them. A hit /// missing what the join needs is an error, not a guess. #[test] -fn qdrant_search_sends_no_filter_and_reads_ids_kinds_and_scores() { +fn qdrant_search_scores_only_the_members_exactly_and_reads_ids_kinds_and_scores() { let name = collection(); let answering = stub(&[( "POST /collections/", 200, r#"{"result":[{"id":"p1","score":0.75,"payload":{"doc_id":"eureka-state:question:Q1","kind":"question"}},{"id":"p2","score":0.5,"payload":{"doc_id":"eureka-state:ruling:R1","kind":"ruling"}}]}"#, )]); - let hits = QdrantIndex::new(&answering.url).search(&name, &[0.5, 1.0, 2.0, 3.0], 12, TIME).unwrap(); + let hits = QdrantIndex::new(&answering.url).search(&name, &[0.5, 1.0, 2.0, 3.0], &["p1".to_owned(), "p2".to_owned()], 12, TIME).unwrap(); assert_eq!( hits, [ @@ -1095,20 +1101,32 @@ fn qdrant_search_sends_no_filter_and_reads_ids_kinds_and_scores() { ); let seen = answering.seen.lock().unwrap(); assert_eq!((seen[0].0.as_str(), seen[0].1.as_str()), ("POST", format!("/collections/{name}/points/search").as_str())); - for expected in ["\"vector\":[0.5,1.0,2.0,3.0]", "\"limit\":12", "\"with_vector\":false"] { - assert!(seen[0].2.contains(expected), "{expected} in {}", seen[0].2); - } - assert!(!seen[0].2.contains("filter"), "no predicate is lowered into the index: {}", seen[0].2); + let body: serde_json::Value = serde_json::from_str(&seen[0].2).expect("the request body is JSON"); + assert_eq!(body["vector"], serde_json::json!([0.5, 1.0, 2.0, 3.0])); + assert_eq!(body["filter"], serde_json::json!({ "must": [{ "has_id": ["p1", "p2"] }] }), "members are included, by id alone"); + assert_eq!(body["params"], serde_json::json!({ "exact": true }), "exact search, under params"); + assert_eq!(body["limit"], 12); + assert_eq!(body["with_vector"], false); + assert!(body.get("exact").is_none(), "a top-level exact is ignored by Qdrant: {body}"); drop(seen); + // The limit is the caller's by value, across the range top_k takes: no + // constant, clamp or cap sits between the daemon and Qdrant. + for limit in [1, 12, 200] { + let echoing = stub(&[("POST /collections/", 200, r#"{"result":[]}"#)]); + QdrantIndex::new(&echoing.url).search(&name, &[1.0], &["p1".to_owned()], limit, TIME).unwrap(); + let sent: serde_json::Value = serde_json::from_str(&echoing.seen.lock().unwrap()[0].2).expect("the request body is JSON"); + assert_eq!(sent["limit"], limit, "the limit asked for is the limit sent"); + } + let nameless = stub(&[("POST /collections/", 200, r#"{"result":[{"id":"p","score":0.5,"payload":{"kind":"question"}}]}"#)]); - assert!(QdrantIndex::new(&nameless.url).search(&name, &[1.0], 1, TIME).is_err()); + assert!(QdrantIndex::new(&nameless.url).search(&name, &[1.0], &[], 1, TIME).is_err()); let failing = stub(&[("POST /collections/", 404, r#"{"status":{"error":"not found"}}"#)]); - assert!(QdrantIndex::new(&failing.url).search(&name, &[1.0], 1, TIME).is_err()); + assert!(QdrantIndex::new(&failing.url).search(&name, &[1.0], &[], 1, TIME).is_err()); } fn queued(ticket: u64, text: &str) -> Queued { - Queued { ticket: SearchTicket(ticket), text: text.into(), top_k: 3, due: Instant::now() + SEARCH_DEADLINE } + Queued { ticket: SearchTicket(ticket), text: text.into(), top_k: 3, among: vec![], due: Instant::now() + SEARCH_DEADLINE } } fn shared_with( @@ -1199,7 +1217,7 @@ type Answers = Vec<(SearchTicket, Result)>; /// The sink's own calls, named: the trait is generic over the store and the /// worker does not care which. fn ask_sink(sink: &mut WorkerSink, text: &str, top_k: u32) -> Result { - >::search(sink, text, top_k) + >::search(sink, text, top_k, vec![]) } fn collected(sink: &mut WorkerSink) -> Answers { @@ -1278,7 +1296,7 @@ fn a_search_is_refused_only_for_a_rebuild_or_a_failure() { let (lost_embedder, lost_index) = pair(); let mut rebuilding_now = rebuilding(&lost_embedder, &lost_index, &entries); let searches = lost_index.state.lock().unwrap().searches.len(); - let refusal = format!("{:#}", rebuilding_now.search("anything", 3, SEARCH_DEADLINE).unwrap_err()); + let refusal = format!("{:#}", rebuilding_now.search("anything", 3, &[], SEARCH_DEADLINE).unwrap_err()); assert!(refusal.contains("being rebuilt") && refusal.contains("pending: 4"), "{refusal}"); assert_eq!(lost_index.state.lock().unwrap().searches.len(), searches, "the rebuilt collection was not asked"); assert!(embedder.texts().is_empty() && index.state.lock().unwrap().searches.is_empty()); @@ -1287,7 +1305,7 @@ fn a_search_is_refused_only_for_a_rebuild_or_a_failure() { let mut lagging = projected(&embedder, &index, &entries[..3]); lagging.want(entries[3..].iter().cloned()); assert_eq!(lagging.progress_status(), IndexStatus::Behind { pending: 1 }); - assert!(lagging.search("anything", 3, SEARCH_DEADLINE).is_ok()); + assert!(lagging.search("anything", 3, &[], SEARCH_DEADLINE).is_ok()); } /// A search that fails for its own reasons (the vector store errs, the @@ -1309,11 +1327,11 @@ fn a_failed_search_fails_that_search_alone() { let mut projector = projected(&embedder, &index, &entries); let listed = index.state.lock().unwrap().listed; fault(&embedder, &index, true); - assert!(projector.search("anything", 3, Duration::from_millis(50)).is_err(), "{what}"); + assert!(projector.search("anything", 3, &[], Duration::from_millis(50)).is_err(), "{what}"); assert_eq!(projector.progress_status(), IndexStatus::Current, "{what}"); assert!(!projector.stale(), "{what}"); fault(&embedder, &index, false); - assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_ok(), "{what}: the next search is answered"); + assert!(projector.search("anything", 3, &[], SEARCH_DEADLINE).is_ok(), "{what}: the next search is answered"); assert_eq!(index.state.lock().unwrap().listed, listed, "{what}: the collection was not listed again"); } @@ -1344,7 +1362,7 @@ fn a_model_changed_refusal_names_the_state_and_the_pending_count() { let mut projector = projected(&embedder, &index, &entries[..3]); projector.want(entries[3..].iter().cloned()); embedder.state.lock().unwrap().identity.digest = "d2".into(); - let refusal = format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err()); + let refusal = format!("{:#}", projector.search("anything", 3, &[], SEARCH_DEADLINE).unwrap_err()); assert!(refusal.contains("model changed") && refusal.contains("being rebuilt") && refusal.contains("pending: 1"), "{refusal}"); } @@ -1374,7 +1392,7 @@ fn a_search_re_reads_the_collections_label() { let mut projector = projected(&embedder, &index, &entries); change(index.state.lock().unwrap().collections.get_mut(&collection()).unwrap()); let embedded = embedder.texts().len(); - let refusal = format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err()); + let refusal = format!("{:#}", projector.search("anything", 3, &[], SEARCH_DEADLINE).unwrap_err()); assert!(refusal.contains("no longer the one this mind's index was built as"), "{what}: {refusal}"); assert_eq!(embedder.texts().len(), embedded, "{what}: nothing was embedded"); assert!(index.state.lock().unwrap().searches.is_empty(), "{what}"); @@ -1383,7 +1401,7 @@ fn a_search_re_reads_the_collections_label() { let (embedder, index) = pair(); let mut projector = projected(&embedder, &index, &entries); index.state.lock().unwrap().collections.clear(); - assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_err(), "no collection"); + assert!(projector.search("anything", 3, &[], SEARCH_DEADLINE).is_err(), "no collection"); assert!(projector.stale()); } @@ -1400,7 +1418,7 @@ fn a_model_that_changes_while_the_query_is_embedded_is_refused() { let mut projector = projected(&embedder, &index, &entries); gate.shut_again(); let searching = std::thread::spawn(move || { - let found = projector.search("anything", 3, SEARCH_DEADLINE); + let found = projector.search("anything", 3, &[], SEARCH_DEADLINE); (projector, found) }); gate.wait_arrived(1); @@ -1431,7 +1449,8 @@ fn every_search_has_its_own_ticket_and_its_answer_comes_back_exactly_once() { )]; let mut tickets = Vec::new(); for text in ["a", "b", "c"] { - let ticket = ask_sink(&mut sink, text, 3).unwrap(); + let among = vec!["eureka-state:question:Q1".to_owned()]; + let ticket = >::search(&mut sink, text, 3, among).unwrap(); assert_eq!(answers(&mut sink, 1), vec![(ticket, Ok(expected.clone()))]); assert!(collected(&mut sink).is_empty(), "an answer is handed over once"); tickets.push(ticket); @@ -1767,7 +1786,7 @@ fn a_search_reads_an_admission_acknowledged_before_it_was_asked() { #[test] fn a_collection_made_again_refuses_searches_until_it_is_refilled() { let refused = |projector: &mut Projector, pending: u32| { - let why = format!("{:#}", projector.search("anything", 3, SEARCH_DEADLINE).unwrap_err()); + let why = format!("{:#}", projector.search("anything", 3, &[], SEARCH_DEADLINE).unwrap_err()); assert!(why.contains("being rebuilt") && why.contains(&format!("pending: {pending}")), "{why}"); }; let entries = entries_of(BATCH + 8); @@ -1787,7 +1806,7 @@ fn a_collection_made_again_refuses_searches_until_it_is_refilled() { refused(&mut projector, 8); assert_eq!(projector.advance().unwrap(), Advance::Progressed); assert_eq!(projector.progress_status(), IndexStatus::Current); - assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_ok(), "answered once refilled"); + assert!(projector.search("anything", 3, &[], SEARCH_DEADLINE).is_ok(), "answered once refilled"); assert_eq!(index.state.lock().unwrap().held_when_searched, [40]); projector.want(entries_of(1).into_iter().map(|mut entry| { entry.id.id.0.push_str("-later"); @@ -1805,7 +1824,7 @@ fn a_collection_made_again_refuses_searches_until_it_is_refilled() { assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 8 }); refused(&mut projector, 8); drain(&mut projector); - assert!(projector.search("anything", 3, SEARCH_DEADLINE).is_ok()); + assert!(projector.search("anything", 3, &[], SEARCH_DEADLINE).is_ok()); } /// Through the worker: a collection lost while the worker runs is made again, @@ -1922,7 +1941,7 @@ fn a_later_calls_bound_shrinks_as_the_search_spends_its_time() { state.identity_bounds.clear(); } index.state.lock().unwrap().bounds.clear(); - projector.search("anything", 3, within).unwrap(); + projector.search("anything", 3, &[], within).unwrap(); let embedder = embedder.state.lock().unwrap(); let index = index.state.lock().unwrap(); diff --git a/crates/huginn-daemon/src/semantic_tests.rs b/crates/huginn-daemon/src/semantic_tests.rs index 11da3f5..4d46b90 100644 --- a/crates/huginn-daemon/src/semantic_tests.rs +++ b/crates/huginn-daemon/src/semantic_tests.rs @@ -268,7 +268,8 @@ fn healthy() -> Harness { /// hold, all scoring above the two that stand, and only the two that stand come /// back, in score order. The selection's other predicates still cut the /// candidates, and the embedder saw the query behind its instruction while the -/// index was asked for an oversample. +/// index was asked for exactly top_k among the members, and `matched` counts +/// the members. #[test] fn a_semantic_query_returns_only_documents_the_mind_holds_and_the_selection_admits() { let harness = healthy(); @@ -286,7 +287,9 @@ fn a_semantic_query_returns_only_documents_the_mind_holds_and_the_selection_admi let standing_only = ask(&mut client, "m-1", &semantic(standing(), 5)); assert_eq!(ids(&standing_only), vec![id("ruling", "R4"), id("question", "Q2")]); let HuginnMindResponse::Query(page) = &standing_only else { unreachable!() }; - assert_eq!((page.matched, page.next.as_ref()), (2, None)); + let members = ids(&ask(&mut client, "m-0", &plain_of(standing()))).len(); + assert!(members > 2, "the mind has more in force than the two hits: {members}"); + assert_eq!((page.matched as usize, page.next.as_ref()), (members, None), "matched counts the members, not the hits"); let rulings = Selection { schemas: Some(vec![PipelineKind::Ruling.type_id().into()]), ..standing() }; assert_eq!(ids(&ask(&mut client, "m-2", &semantic(rulings, 5))), vec![id("ruling", "R4")], "the selection's schemas still apply"); @@ -298,7 +301,93 @@ fn a_semantic_query_returns_only_documents_the_mind_holds_and_the_selection_admi let embedded = harness.embedder.texts(); assert!(embedded.iter().any(|text| text.starts_with("Instruct: ") && text.ends_with("\nQuery: who owns the state")), "{embedded:?}"); let searches = harness.index.state.lock().unwrap().searches.clone(); - assert!(searches.iter().all(|(name, _, limit)| *name == collection_name(&slug(INSTANCE)) && *limit == 20), "{searches:?}"); + assert!(searches.iter().all(|(name, _, limit)| *name == collection_name(&slug(INSTANCE)) && *limit == 5), "top_k, with no oversample: {searches:?}"); +} + +fn plain_of(selection: Selection) -> HuginnMindRequest { + HuginnMindRequest::Query { instance: slug(INSTANCE), selection, semantic: None } +} + +/// Thirty rulings stand out of force (superseded by K0) and five stand, and the +/// index ranks every one that does not stand above every one that does. The +/// members reach the index, so the nearest top_k of them are the answer: +/// top_k 4 answers four rows, where an index window cut before the mind judged +/// would have answered none. +#[test] +fn a_narrow_selection_gets_top_k_members_when_non_members_score_higher() { + let gone = |n: u32| id("ruling", &format!("N{n}")); + let kept = |n: u32| id("ruling", &format!("K{n}")); + let mut seed = Vec::new(); + seed.push(campaign_seed()); + seed.push((0..5).map(|n| ruling(&format!("K{n}"), None)).collect()); + seed.push((0..30).map(|n| ruling(&format!("N{n}"), None)).collect()); + seed.push( + (0..30) + .map(|n| { + resolution( + PipelineKind::Ruling, + &gone(n), + ResolutionOutcome::Superseded { by: vec![PipelineRef { kind: PipelineKind::Ruling, id: Short(kept(0)) }] }, + ) + }) + .collect(), + ); + let (embedder, index) = (FakeEmbedder::new("d1"), FakeIndex::default()); + let harness = Harness::start(seed, embedder, index, Duration::from_secs(30)); + let mut client = harness.client(); + until_index(&mut client, "Current", |status| *status == IndexStatus::Current); + let mut hits: Vec = (0..30).map(|n| hit("ruling", gone(n), 0.9 - n as f32 * 0.001)).collect(); + hits.extend((0..5).map(|n| hit("ruling", kept(n), 0.5 - n as f32 * 0.1))); + harness.index.state.lock().unwrap().hits = hits; + + let rulings = Selection { schemas: Some(vec![PipelineKind::Ruling.type_id().into()]), ..standing() }; + let answer = ask(&mut client, "m-1", &semantic(rulings, 4)); + assert_eq!(ids(&answer), (0..4).map(kept).collect::>(), "the four nearest members, none of the nearer non-members"); + let HuginnMindResponse::Query(page) = &answer else { unreachable!() }; + assert_eq!(page.matched, 5); + let scores = page.scores.clone().expect("a semantic page carries scores"); + assert_eq!(scores.len(), 4); + assert!(scores.windows(2).all(|pair| pair[0] >= pair[1]), "best first: {scores:?}"); + assert!((scores[0] - 0.5).abs() < 1e-6, "the score is the index's: {scores:?}"); +} + +/// The daemon forwards the caller's top_k to the index as asked, however large: +/// twenty members stand, top_k 15 reaches the index as 15 and answers fifteen +/// rows, and top_k 200 reaches it as 200 and answers all twenty. +#[test] +fn a_large_top_k_reaches_the_index_whole_and_answers_that_many_members() { + let kept = |n: u32| id("ruling", &format!("K{n:02}")); + let seed = vec![campaign_seed(), (0..20).map(|n| ruling(&format!("K{n:02}"), None)).collect()]; + let harness = Harness::start(seed, FakeEmbedder::new("d1"), FakeIndex::default(), Duration::from_secs(30)); + let mut client = harness.client(); + until_index(&mut client, "Current", |status| *status == IndexStatus::Current); + harness.index.state.lock().unwrap().hits = (0..20).map(|n| hit("ruling", kept(n), 0.9 - n as f32 * 0.01)).collect(); + let rulings = Selection { schemas: Some(vec![PipelineKind::Ruling.type_id().into()]), ..standing() }; + + let fifteen = ask(&mut client, "m-1", &semantic(rulings.clone(), 15)); + assert_eq!(ids(&fifteen), (0..15).map(kept).collect::>()); + let all = ask(&mut client, "m-2", &semantic(rulings, 200)); + assert_eq!(ids(&all), (0..20).map(kept).collect::>()); + let limits: Vec = harness.index.state.lock().unwrap().searches.iter().map(|(_, _, limit)| *limit).collect(); + assert_eq!(limits, vec![15, 200], "the index is asked for the caller's top_k, not a clamp of it"); +} + +/// A selection that matches nothing is an empty page at once: the query is not +/// embedded and the index is not asked. +#[test] +fn an_empty_membership_is_an_empty_page_and_the_index_is_never_asked() { + let harness = healthy(); + let mut client = harness.client(); + until_index(&mut client, "Current", |status| *status == IndexStatus::Current); + let embedded = harness.embedder.texts().len(); + harness.index.state.lock().unwrap().hits = vec![hit("question", id("question", "Q1"), 0.9)]; + + let findings = Selection { schemas: Some(vec![PipelineKind::Finding.type_id().into()]), ..standing() }; + let answer = ask(&mut client, "m-1", &semantic(findings, 5)); + let HuginnMindResponse::Query(page) = &answer else { panic!("{answer:?}") }; + assert_eq!((page.matched, ids(&answer).len(), page.scores.clone()), (0, 0, Some(vec![]))); + assert_eq!(harness.embedder.texts().len(), embedded, "nothing was embedded"); + assert!(harness.index.state.lock().unwrap().searches.is_empty(), "the index was not asked"); } /// A cursor cannot resume a ranked answer and a ranked answer has no order to diff --git a/crates/huginn-daemon/tests/live_index_smoke.rs b/crates/huginn-daemon/tests/live_index_smoke.rs index 85fabc2..c6a2c9e 100644 --- a/crates/huginn-daemon/tests/live_index_smoke.rs +++ b/crates/huginn-daemon/tests/live_index_smoke.rs @@ -46,9 +46,13 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() { let mut embedder = OllamaEmbedder::new(&ollama, &model); let identity = embedder.model_identity(Duration::from_secs(60)).expect("the model is listed"); println!("identity: {identity:?}"); - let texts = ["The organ owns the mind.".to_string(), "A recipe for lentil soup.".to_string()]; - let vectors = embedder.embed(&texts, Duration::from_secs(60)).expect("two texts embed"); - assert_eq!(vectors.len(), 2); + let texts = [ + "The organ owns the mind.".to_string(), + "A recipe for lentil soup.".to_string(), + "How to keep a sourdough starter alive.".to_string(), + ]; + let vectors = embedder.embed(&texts, Duration::from_secs(60)).expect("three texts embed"); + assert_eq!(vectors.len(), 3); assert!(vectors.iter().all(|vector| vector.len() == identity.dimensions as usize)); let meta = CollectionMeta { @@ -97,10 +101,17 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() { let asked = format!("Instruct: {QUERY_INSTRUCTION} Query: Who is responsible for keeping the memory?"); let query = embedder.embed(&[asked], Duration::from_secs(60)).expect("the query embeds").remove(0); - let hits = store.search(name, &query, 2, Duration::from_secs(60)).unwrap(); + let among: Vec = points.iter().map(|point| point.id.clone()).collect(); + let hits = store.search(name, &query, &among, 2, Duration::from_secs(60)).unwrap(); assert_eq!(hits.len(), 2, "{hits:?}"); assert_eq!((hits[0].doc_id.as_str(), hits[0].kind.as_str()), ("smoke:doc:0", "question"), "{hits:?}"); assert!(hits[0].score > hits[1].score, "{hits:?}"); + // The members are a strict subset that leaves out the nearest point: + // the filter must exclude it, not merely rank it. + let subset: Vec = points[1..].iter().map(|point| point.id.clone()).collect(); + let hits = store.search(name, &query, &subset, 3, Duration::from_secs(60)).unwrap(); + let docs: BTreeSet<&str> = hits.iter().map(|hit| hit.doc_id.as_str()).collect(); + assert_eq!(docs, BTreeSet::from(["smoke:doc:1", "smoke:doc:2"]), "only the members are scored: {hits:?}"); println!("{name}: described, created, upserted {} points, queried, searched: {hits:?}", points.len()); } })); diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index aef0136..63a64bc 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -48,7 +48,7 @@ pub use mind::{HuginnMindEpoch, Mind}; pub use persona::{PersonaPutOutcome, PersonaStateView}; pub use query::{ AdmissionFacts, PipelineDocumentSummary, PipelineDocumentView, PipelineEdge, PipelineFacts, PipelinePageItems, - PipelineSelectionPage, PipelineStatus, PipelineStatusSummary, Ranked, SUMMARY_MAX_BYTES, SemanticQuery, + Members, PipelineSelectionPage, PipelineStatus, PipelineStatusSummary, Ranked, SUMMARY_MAX_BYTES, SemanticQuery, }; pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, PipelineProvenance, RECEIPT_SCHEMA_VERSION}; pub use refusal::{MindRefusal, RunId}; diff --git a/crates/huginn-mind/src/query.rs b/crates/huginn-mind/src/query.rs index 8debedb..1f1945b 100644 --- a/crates/huginn-mind/src/query.rs +++ b/crates/huginn-mind/src/query.rs @@ -20,7 +20,7 @@ //! scores, and this module joins them back through the mind and lets the //! selection decide. No clock, no store handle, no network. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use cultnet_rs::{ Cursor, EdgeAnchor, EdgeMatch, Evaluation, LIMIT_MAX, LIMIT_MIN, PROJECTION_DOCUMENT, Row, Selection, select, validate, @@ -77,7 +77,9 @@ pub struct PipelineDocumentView { /// A semantic query: the text to find documents near, and how many. The index /// owns which fields of each kind are text (`index_text`), so no substring -/// filter lives here to become a second answer to that question. +/// filter lives here to become a second answer to that question. `top_k` is +/// the most rows answered, all of them members of the selection: the +/// selection decides membership before the index is asked. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] pub struct SemanticQuery { pub text: Line, @@ -272,15 +274,29 @@ pub struct PipelineEdge { } /// One page of a selection. `matched` is the count before paging, `as_of` the -/// snapshot the page and every cursor of its walk are exact as of, and `edges` -/// is present exactly when the selection follows a hop. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +/// snapshot the page and every cursor of its walk are exact as of, `edges` +/// is present exactly when the selection follows a hop, and `scores` exactly +/// when the answer is semantic: one cosine similarity per item, in order +/// (higher is nearer), for display and thresholding. It is never stored. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)] pub struct PipelineSelectionPage { pub matched: u32, pub as_of: u64, pub next: Option, pub items: PipelinePageItems, pub edges: Option>, + pub scores: Option>, +} + +/// The selection's whole match set at one snapshot, in the selection's order: +/// what a cursor walk of `query` over the same selection returns. Local to +/// the process, like `Ranked`; it is never on the wire. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Members { + pub as_of: u64, + pub ids: Vec, + /// How many select pages the walk took: the cost of the membership. + pub pages: u32, } /// A ranked semantic answer: the page, and the hits the index named that the @@ -468,7 +484,7 @@ impl Mind { }; let rows = Reader::at(self, as_of)?.rows()?; let evaluation = select(&Vocabulary, &rows, &selection, as_of, self.cursor_key())?; - Self::page(&selection, evaluation, as_of) + Self::page(&selection, evaluation, as_of, None) } /// What a semantic query must satisfy before anything is embedded: the @@ -503,54 +519,87 @@ impl Mind { refuse_values(selection) } - /// Whether a semantic query may be started, decided before the index is - /// asked anything. `rank` runs the same check again over the same inputs. - pub fn check_semantic(&self, selection: &Selection, semantic: &SemanticQuery) -> Result<(), MindRefusal> { - Self::semantic_selection(selection, semantic).map(drop) + /// The members of a semantic query: every document the selection matches + /// at the head, decided before the index is asked anything. + /// + /// The selection passes the doors `semantic_selection` names, then is + /// walked by cursor at one snapshot, a page of LIMIT_MAX at a time, until it has no + /// next page, so `hops` and `in_force` decide membership exactly as + /// `query` does. The index never decides who is a member; it scores the + /// ids this returns. + pub fn members(&self, selection: &Selection, semantic: &SemanticQuery) -> Result { + let selection = Self::semantic_selection(selection, semantic)?; + let head = receipt::head(self)?; + let rows = Reader::at(self, head)?.rows()?; + // The walk is built here, field by field, from the caller's filters + // alone and never spread from the caller's selection: the page size + // is ours (the caller's `limit` shapes an answer, not the walk's cost: + // one select per member at limit 1), from the first page, in the + // one order. + let Selection { schemas, keys, fields, cites, cited, .. } = selection; + let mut walk = Selection { + schemas, + keys, + fields, + cites, + cited, + limit: Some(LIMIT_MAX), + ..Selection::default() + }; + let mut ids = Vec::new(); + let mut pages = 0; + loop { + let evaluation = select(&Vocabulary, &rows, &walk, head, self.cursor_key())?; + pages += 1; + ids.extend(evaluation.rows.iter().map(ref_of)); + match evaluation.next_cursor { + Some(next) => walk.cursor = Some(next), + None => return Ok(Members { as_of: head, ids, pages }), + } + } } - /// The page a semantic query answers, from the candidates the index gave, - /// with the hits the mind holds no document for. + /// The page a semantic query answers, from the hits the index scored + /// among the members, with the hits the mind refused. /// - /// The index supplies ids and scores and nothing else. A hit is a - /// candidate: it is dropped unless the mind holds a document of that kind - /// and id at the head, and what remains is decided by the selection, run - /// by the substrate over every row the mind has, its `keys` narrowed to - /// the candidates. Narrowing `keys` rather than the rows keeps the - /// evaluator's universe whole, so `cited` and `in_force` mean what they - /// mean in `query`. Nothing about a document's standing is read from the - /// index; `in_force` is the caller's to ask for and the mind's to derive. - /// - /// This is the one place the mind judges whether it holds a hit, in the - /// one read of its rows that ranking needs: `Ranked::dropped` returns the - /// hits it found no document for, for the caller to report. + /// The mind decides membership first (`members`); the index only scores + /// the ids it was handed, so the answer is exact as of `members.as_of`, + /// the snapshot the members were taken at, and not the head. The index + /// supplies ids and scores and nothing else. A hit is dropped, and + /// reported in `Ranked::dropped`, unless the mind holds a document of + /// that kind and id at that snapshot and the id is one of `members.ids`. + /// What remains is evaluated by the substrate over every row the mind + /// had, its `keys` narrowed to the hits. Narrowing `keys` rather than the + /// rows keeps the evaluator's universe whole, so `cited` and `in_force` + /// mean what they mean in `query`. Nothing about a document's standing is + /// read from the index. /// /// The order is the one thing the substrate cannot know: score, highest /// first, ties by ordinal. The page holds at most the selection's limit - /// and at most `top_k` documents, `matched` counts every candidate the - /// selection passed, and `next` is never set. The substrate answers at - /// most `LIMIT_MAX` rows and cuts by ordinal, so the candidates are cut - /// to the best `LIMIT_MAX` by score first, among the documents the mind - /// holds: an id the mind does not hold never takes the place of one it - /// does. + /// and at most `top_k` documents, `matched` counts the members, `scores` + /// holds one score per kept row, in row order, and `next` is never set. pub fn rank( &self, selection: &Selection, semantic: &SemanticQuery, + members: &Members, hits: &[(PipelineRef, f32)], ) -> Result { let selection = Self::semantic_selection(selection, semantic)?; - let head = receipt::head(self)?; - let rows = Reader::at(self, head)?.rows()?; + let as_of = members.as_of; + let rows = Reader::at(self, as_of)?.rows()?; let held: BTreeMap<(PipelineKind, &str), i64> = rows.iter().map(|row| ((row.kind, row.key.as_str()), row.ordinal())).collect(); + let member: BTreeSet<(PipelineKind, &str)> = + members.ids.iter().map(|id| (id.kind, id.id.0.as_str())).collect(); let mut candidates: Vec<(&str, f32, i64)> = Vec::new(); let mut dropped = Vec::new(); for (hit, score) in hits { - match held.get(&(hit.kind, hit.id.0.as_str())) { - Some(&ordinal) => candidates.push((hit.id.0.as_str(), *score, ordinal)), - None => dropped.push(hit.clone()), + let id = (hit.kind, hit.id.0.as_str()); + match held.get(&id) { + Some(&ordinal) if member.contains(&id) => candidates.push((id.1, *score, ordinal)), + _ => dropped.push(hit.clone()), } } candidates.sort_by(|a, b| b.1.total_cmp(&a.1).then(a.2.cmp(&b.2))); @@ -569,12 +618,13 @@ impl Mind { Evaluation { rows: Vec::new(), matched: 0, edges: Vec::new(), next_cursor: None } } else { let narrowed = Selection { keys: Some(keys), limit: Some(LIMIT_MAX), ..selection.clone() }; - select(&Vocabulary, &rows, &narrowed, head, self.cursor_key())? + select(&Vocabulary, &rows, &narrowed, as_of, self.cursor_key())? }; let score = |row: &SelectionRow| scores.get(row.key.as_str()).copied().unwrap_or(f32::NEG_INFINITY); evaluation.rows.sort_by(|a, b| score(b).total_cmp(&score(a)).then(a.ordinal().cmp(&b.ordinal()))); evaluation.rows.truncate(limit); evaluation.next_cursor = None; + let row_scores: Vec = evaluation.rows.iter().map(score).collect(); let position = |kind: PipelineKind, key: &str| evaluation.rows.iter().position(|row| row.kind == kind && row.key == key); let mut kept: Vec<(usize, EdgeMatch)> = std::mem::take(&mut evaluation.edges) @@ -589,7 +639,8 @@ impl Mind { .collect(); kept.sort_by_key(|(at, _)| *at); evaluation.edges = kept.into_iter().map(|(_, edge)| edge).collect(); - Ok(Ranked { page: Self::page(&selection, evaluation, head)?, dropped }) + evaluation.matched = u32::try_from(members.ids.len()).unwrap_or(u32::MAX); + Ok(Ranked { page: Self::page(&selection, evaluation, as_of, Some(row_scores))?, dropped }) } /// The projection of an evaluation: a header or a whole view per row, and @@ -598,6 +649,7 @@ impl Mind { selection: &Selection, evaluation: Evaluation, as_of: u64, + scores: Option>, ) -> Result { let items = if selection.projection == PROJECTION_DOCUMENT { PipelinePageItems::Documents(evaluation.rows.iter().map(|row| row.view.clone()).collect()) @@ -616,7 +668,7 @@ impl Mind { } else { None }; - Ok(PipelineSelectionPage { matched: evaluation.matched, as_of, next: evaluation.next_cursor, items, edges }) + Ok(PipelineSelectionPage { matched: evaluation.matched, as_of, next: evaluation.next_cursor, items, edges, scores }) } } @@ -1594,6 +1646,13 @@ mod tests { with(selection, any_of("in_force", &["true"])) } + /// The members of the selection from the mind, then the page ranked from + /// the hits: the two steps the daemon takes around the index. + fn rank_all(mind: &Mind, selection: &Selection, semantic: &SemanticQuery, hits: &[(PipelineRef, f32)]) -> Ranked { + let members = mind.members(selection, semantic).unwrap(); + mind.rank(selection, semantic, &members, hits).unwrap() + } + /// A mind with every way a document stops being in force: Q1 answered /// (resolved), Q2 answered and the answer withdrawn (its resolution is /// resolved, the question in force again), R3 superseded by R4. @@ -1637,7 +1696,7 @@ mod tests { hit(K::Ruling, &shelf.r4, 0.5), hit(K::Question, &shelf.q2, 0.4), ]; - let ranked = shelf.mind.rank(&Selection::default(), &asking(10), &hits).unwrap(); + let ranked = rank_all(&shelf.mind, &Selection::default(), &asking(10), &hits); assert_eq!( ranked.dropped, vec![r(K::Question, &id("question", "Q99")), r(K::Question, &shelf.r4)], @@ -1646,16 +1705,17 @@ mod tests { let everything = ranked.page; assert_eq!(ids(&everything), vec![shelf.q1.clone(), shelf.withdrawn_resolution.clone(), shelf.r3.clone(), shelf.r4.clone(), shelf.q2.clone()], "without in_force the selection has not asked about standing, and the unheld and mis-kinded hits are gone"); - assert_eq!(everything.matched, 5); + assert_eq!(everything.matched, plain_matched(&shelf.mind, &Selection::default()), "matched counts the whole mind, not the five hits"); - let page = shelf.mind.rank(&standing(Selection::default()), &asking(10), &hits).unwrap().page; + let page = rank_all(&shelf.mind, &standing(Selection::default()), &asking(10), &hits).page; assert_eq!(ids(&page), vec![shelf.r4.clone(), shelf.q2.clone()]); - assert_eq!((page.matched, page.next), (2, None)); + assert_eq!((page.matched, page.next), (plain_matched(&shelf.mind, &standing(Selection::default())), None)); assert_eq!(page.as_of, receipt::head(&shelf.mind).unwrap()); - let none = shelf.mind.rank(&standing(Selection::default()), &asking(10), &[hit(K::Question, &id("question", "Q99"), 1.0)]).unwrap().page; - assert_eq!((none.matched, ids(&none)), (0, Vec::::new()), "an empty candidate set is an empty page, not a refusal"); - assert_eq!(shelf.mind.rank(&standing(Selection::default()), &asking(10), &[]).unwrap().page.matched, 0); + let none = rank_all(&shelf.mind, &standing(Selection::default()), &asking(10), &[hit(K::Question, &id("question", "Q99"), 1.0)]).page; + assert_eq!(ids(&none), Vec::::new(), "an empty candidate set is an empty page, not a refusal"); + assert_eq!(none.matched, plain_matched(&shelf.mind, &standing(Selection::default())), "and the members are still counted"); + assert_eq!(rank_all(&shelf.mind, &standing(Selection::default()), &asking(10), &[]).page.matched, none.matched); } /// The substrate answers 200 rows and cuts by ordinal, so 250 candidates @@ -1673,10 +1733,11 @@ mod tests { let mut hits: Vec<(PipelineRef, f32)> = (0..250).map(|n| hit(K::Question, &key(n), n as f32)).collect(); // Twenty ids the mind holds nothing for, above every real score. hits.extend((0..20).map(|n| hit(K::Question, &id("question", &format!("Gone{n}")), 1000.0 + n as f32))); - let ranked = mind.rank(&Selection::default(), &asking(200), &hits).unwrap(); + let ranked = rank_all(&mind, &Selection::default(), &asking(200), &hits); let expected: Vec = (50..250).rev().map(key).collect(); assert_eq!(ids(&ranked.page), expected, "the 200 best, best first"); - assert_eq!(ranked.page.matched, 200); + assert_eq!(ranked.page.matched, plain_matched(&mind, &Selection::default())); + assert!(ranked.page.matched > 250, "every member, not the 200 answered"); assert_eq!(ranked.dropped.len(), 20); } @@ -1687,17 +1748,17 @@ mod tests { fn ranked_results_are_ordered_by_score_then_ordinal_and_cut_to_the_limit() { let shelf = shelf(); let by_ordinal = [hit(K::Question, &shelf.q2, 0.4), hit(K::Ruling, &shelf.r4, 0.5)]; - assert_eq!(ids(&shelf.mind.rank(&standing(Selection::default()), &asking(10), &by_ordinal).unwrap().page), vec![shelf.r4.clone(), shelf.q2.clone()], + assert_eq!(ids(&rank_all(&shelf.mind, &standing(Selection::default()), &asking(10), &by_ordinal).page), vec![shelf.r4.clone(), shelf.q2.clone()], "Q2 was admitted first, and the better score still comes first"); let tied = [hit(K::Ruling, &shelf.r4, 0.5), hit(K::Question, &shelf.q2, 0.5)]; - assert_eq!(ids(&shelf.mind.rank(&standing(Selection::default()), &asking(10), &tied).unwrap().page), vec![shelf.q2.clone(), shelf.r4.clone()], + assert_eq!(ids(&rank_all(&shelf.mind, &standing(Selection::default()), &asking(10), &tied).page), vec![shelf.q2.clone(), shelf.r4.clone()], "a tie falls to the earlier ordinal"); let limited = Selection { limit: Some(1), ..standing(Selection::default()) }; - let page = shelf.mind.rank(&limited, &asking(10), &by_ordinal).unwrap().page; - assert_eq!((ids(&page), page.matched, page.next), (vec![shelf.r4.clone()], 2, None), "the limit keeps the best, not the earliest"); - let page = shelf.mind.rank(&standing(Selection::default()), &asking(1), &by_ordinal).unwrap().page; - assert_eq!((ids(&page), page.matched), (vec![shelf.r4.clone()], 2), "top_k bounds the page too"); + let page = rank_all(&shelf.mind, &limited, &asking(10), &by_ordinal).page; + assert_eq!((ids(&page), page.matched, page.next), (vec![shelf.r4.clone()], plain_matched(&shelf.mind, &standing(Selection::default())), None), "the limit keeps the best, not the earliest"); + let page = rank_all(&shelf.mind, &standing(Selection::default()), &asking(1), &by_ordinal).page; + assert_eq!((ids(&page), page.matched), (vec![shelf.r4.clone()], plain_matched(&shelf.mind, &standing(Selection::default()))), "top_k bounds the page too"); } /// Membership is the selection's alone: a kind, a key allowlist and a @@ -1708,16 +1769,18 @@ mod tests { fn the_selections_own_predicates_decide_which_candidates_stay() { let shelf = shelf(); let hits = [hit(K::Question, &shelf.q2, 0.4), hit(K::Ruling, &shelf.r4, 0.5)]; - assert_eq!(ids(&shelf.mind.rank(&of_kinds(&[K::Ruling]), &asking(10), &hits).unwrap().page), vec![shelf.r4.clone()]); + assert_eq!(ids(&rank_all(&shelf.mind, &of_kinds(&[K::Ruling]), &asking(10), &hits).page), vec![shelf.r4.clone()]); let keyed = Selection { keys: Some(vec![shelf.q2.clone(), shelf.q1.clone()]), ..Selection::default() }; - assert_eq!(ids(&shelf.mind.rank(&keyed, &asking(10), &hits).unwrap().page), vec![shelf.q2.clone()], "keys intersect the candidates"); + assert_eq!(ids(&rank_all(&shelf.mind, &keyed, &asking(10), &hits).page), vec![shelf.q2.clone()], "keys intersect the candidates"); let disjoint = Selection { keys: Some(vec![shelf.q1.clone()]), ..Selection::default() }; - assert_eq!(shelf.mind.rank(&disjoint, &asking(10), &hits).unwrap().page.matched, 0); + let none = rank_all(&shelf.mind, &disjoint, &asking(10), &hits); + assert_eq!((none.page.matched, ids(&none.page).len()), (1, 0), "Q1 is the one member, and neither hit is it"); + assert_eq!(none.dropped.len(), 2, "hits outside the members are dropped"); let unnamed = Selection { cited: Some(Incoming { role: "subject".into(), exists: false }), ..Selection::default() }; let plain = ids(&shelf.mind.query(&Selection { keys: Some(vec![shelf.q2.clone(), shelf.r4.clone()]), ..unnamed.clone() }).unwrap()); assert_eq!(plain, vec![shelf.r4.clone()], "no resolution names R4 as its subject, and one names Q2"); - assert_eq!(ids(&shelf.mind.rank(&unnamed, &asking(10), &hits).unwrap().page), plain, "the same answer as the plain query over the same rows"); - assert_eq!(ids(&shelf.mind.rank(&Selection { projection: "document".into(), ..of_kinds(&[K::Ruling]) }, &asking(10), &hits).unwrap().page), vec![shelf.r4.clone()]); + assert_eq!(ids(&rank_all(&shelf.mind, &unnamed, &asking(10), &hits).page), plain, "the same answer as the plain query over the same rows"); + assert_eq!(ids(&rank_all(&shelf.mind, &Selection { projection: "document".into(), ..of_kinds(&[K::Ruling]) }, &asking(10), &hits).page), vec![shelf.r4.clone()]); } /// A citation hop is answered over the whole mind, and its edges follow the @@ -1729,14 +1792,14 @@ mod tests { let shelf = shelf(); let named = Selection { cited: Some(Incoming { role: "subject".into(), exists: true }), ..of_kinds(&[K::Question]) }; let hits = [hit(K::Question, &shelf.q2, 0.9), hit(K::Question, &shelf.q1, 0.4)]; - let both = shelf.mind.rank(&named, &asking(10), &hits).unwrap().page; + let both = rank_all(&shelf.mind, &named, &asking(10), &hits).page; assert_eq!(ids(&both), vec![shelf.q2.clone(), shelf.q1.clone()], "Q1 was admitted first and scored lower"); let edges = both.edges.expect("a hop carries its edges"); let targets: Vec<&str> = edges.iter().map(|edge| edge.to.id.0.as_str()).collect(); assert_eq!(targets, [shelf.q2.as_str(), shelf.q1.as_str()], "the edges follow the page order"); assert!(edges.iter().all(|edge| edge.from.kind == K::Resolution && edge.role == CitationRole::Subject)); - let limited = shelf.mind.rank(&Selection { limit: Some(1), ..named.clone() }, &asking(10), &hits).unwrap().page; + let limited = rank_all(&shelf.mind, &Selection { limit: Some(1), ..named.clone() }, &asking(10), &hits).page; assert_eq!(ids(&limited), vec![shelf.q2.clone()]); let kept = limited.edges.unwrap(); assert_eq!(kept.len(), 1, "the dropped row's edge is dropped with it"); @@ -1744,41 +1807,238 @@ mod tests { assert_eq!(kept[0].from.id.0, id("resolution", "question.Q2.n1")); let empty_cursor = Selection { cursor: Some(String::new()), ..standing(Selection::default()) }; - let with = shelf.mind.rank(&empty_cursor, &asking(10), &[hit(K::Ruling, &shelf.r4, 0.5)]).unwrap().page; + let with = rank_all(&shelf.mind, &empty_cursor, &asking(10), &[hit(K::Ruling, &shelf.r4, 0.5)]).page; assert_eq!(ids(&with), vec![shelf.r4.clone()]); } /// A ranked answer cannot be resumed, so a cursor is refused by name, and - /// so is a query that cannot be asked, before any hit is looked at. + /// so is a query that cannot be asked: `members` refuses before the index + /// is asked anything and before any hit is looked at. #[test] fn a_semantic_query_that_cannot_be_ranked_is_refused_by_name() { let shelf = shelf(); - let hits = [hit(K::Ruling, &shelf.r4, 0.5)]; let with_cursor = Selection { cursor: Some("anything".into()), ..Selection::default() }; fn on(field: &str, refusal: &MindRefusal) -> bool { matches!(refusal, MindRefusal::SelectionInvalid { field: f, .. } if f == field) } - for refusal in [ - shelf.mind.rank(&with_cursor, &asking(10), &hits).unwrap_err(), - shelf.mind.check_semantic(&with_cursor, &asking(10)).unwrap_err(), - ] { - assert!(on("cursor", &refusal), "{refusal:?}"); - } + let refusal = shelf.mind.members(&with_cursor, &asking(10)).unwrap_err(); + assert!(on("cursor", &refusal), "{refusal:?}"); let reversed = Selection { descending: true, ..Selection::default() }; - for refusal in [ - shelf.mind.rank(&reversed, &asking(10), &hits).unwrap_err(), - shelf.mind.check_semantic(&reversed, &asking(10)).unwrap_err(), - ] { - assert!(on("descending", &refusal), "a ranked answer has no order to reverse: {refusal:?}"); - } - assert!(on("semantic.top_k", &shelf.mind.rank(&Selection::default(), &asking(0), &hits).unwrap_err())); - assert!(on("semantic.top_k", &shelf.mind.rank(&Selection::default(), &asking(201), &hits).unwrap_err())); + let refusal = shelf.mind.members(&reversed, &asking(10)).unwrap_err(); + assert!(on("descending", &refusal), "a ranked answer has no order to reverse: {refusal:?}"); + assert!(on("semantic.top_k", &shelf.mind.members(&Selection::default(), &asking(0)).unwrap_err())); + assert!(on("semantic.top_k", &shelf.mind.members(&Selection::default(), &asking(201)).unwrap_err())); let blank = SemanticQuery { text: " ".into(), top_k: 3 }; - assert!(on("semantic.text", &shelf.mind.rank(&Selection::default(), &blank, &hits).unwrap_err())); + assert!(on("semantic.text", &shelf.mind.members(&Selection::default(), &blank).unwrap_err())); let malformed = with(Selection::default(), any_of("root", &["a/../b"])); - assert_eq!(shelf.mind.check_semantic(&malformed, &asking(3)), shelf.mind.query(&malformed).map(drop), "the same door as a plain query"); - assert!(shelf.mind.check_semantic(&standing(Selection::default()), &asking(3)).is_ok()); + assert_eq!( + shelf.mind.members(&malformed, &asking(3)).map(drop), + shelf.mind.query(&malformed).map(drop), + "the same door as a plain query" + ); + assert!(shelf.mind.members(&standing(Selection::default()), &asking(3)).is_ok()); + // rank runs the same doors over the same inputs. + let members = shelf.mind.members(&Selection::default(), &asking(10)).unwrap(); + let refusal = shelf.mind.rank(&with_cursor, &asking(10), &members, &[]).unwrap_err(); + assert!(on("cursor", &refusal), "{refusal:?}"); + } + + /// The matched count a plain query gives the same selection: the figure a + /// semantic answer's `matched` must equal, computed without the semantic + /// path. + fn plain_matched(mind: &Mind, selection: &Selection) -> u32 { + mind.query(selection).unwrap().matched + } + + /// Every id a cursor walk of `query` returns for the selection, in order. + fn walked(mind: &Mind, selection: &Selection) -> Vec { + let mut walk = selection.clone(); + let mut out = Vec::new(); + loop { + let page = mind.query(&walk).unwrap(); + match &page.items { + PipelinePageItems::Headers(headers) => out.extend(headers.iter().map(|header| header.id.clone())), + PipelinePageItems::Documents(views) => out.extend(views.iter().map(|view| view.id.clone())), + } + match page.next { + Some(next) => walk.cursor = Some(next), + None => return out, + } + } + } + + /// The members of a selection are everything it matches, not the first + /// page: past 200 rows, with a citation hop and `in_force`, they are the + /// cursor walk of `query` over the same selection, as of the head. + #[test] + fn members_are_the_whole_match_of_the_selection_past_one_page() { + let mut mind = seeded(); + for start in (0..250).step_by(crate::admission::BATCH_MAX) { + committed(admit(&mut mind, (start..(start + crate::admission::BATCH_MAX as u32).min(250)).map(question_n).collect())); + } + // Forty questions are answered: resolved, and named as a subject. + let answered: Vec = + (0..40).map(|n| answering(&format!("R{n}"), &id("question", &format!("Q{n}")))).collect(); + for batch in answered.chunks(crate::admission::BATCH_MAX) { + committed(admit(&mut mind, batch.to_vec())); + } + let questions = of_kinds(&[K::Question]); + let unnamed = Selection { cited: Some(Incoming { role: "subject".into(), exists: false }), ..questions.clone() }; + let selections = [ + Selection::default(), + questions.clone(), + standing(questions.clone()), + unnamed.clone(), + standing(unnamed), + ]; + for selection in selections { + let members = mind.members(&selection, &asking(5)).unwrap(); + assert_eq!(members.as_of, receipt::head(&mind).unwrap()); + assert_eq!(members.ids, walked(&mind, &selection), "{selection:?}"); + assert_eq!(members.ids.len() as u32, plain_matched(&mind, &selection)); + } + let standing_questions = mind.members(&standing(questions), &asking(5)).unwrap(); + assert_eq!(standing_questions.ids.len(), 210, "250 questions, 40 of them resolved, all past one page"); + // The selection's own limit shortens a page, never the membership. + let limited = Selection { limit: Some(3), ..of_kinds(&[K::Question]) }; + assert_eq!(mind.members(&limited, &asking(5)).unwrap().ids.len(), 250); + } + + /// The walk costs one select per LIMIT_MAX members whatever the caller + /// asked for, through the production `members`: a caller's `limit`, + /// projection or spread never reach it. At limit 1 and document + /// projection over 599 questions the walk takes exactly 3 pages and + /// returns exactly the questions, in the mind's order. + #[test] + fn the_members_walk_takes_its_page_size_from_the_mind_not_the_caller() { + let mut mind = seeded(); + for start in (0..599u32).step_by(crate::admission::BATCH_MAX) { + committed(admit(&mut mind, (start..(start + crate::admission::BATCH_MAX as u32).min(599)).map(question_n).collect())); + } + let caller = Selection { limit: Some(1), projection: PROJECTION_DOCUMENT.into(), ..of_kinds(&[K::Question]) }; + let members = mind.members(&caller, &asking(5)).unwrap(); + // The one order: admission order (a batch is one ordinal), then key. + let mut numbered: Vec = (0..599).collect(); + numbered.sort_by_key(|n| (n / crate::admission::BATCH_MAX as u32, format!("Q{n}"))); + let expected: Vec = numbered.iter().map(|n| r(K::Question, &id("question", &format!("Q{n}")))).collect(); + assert_eq!(members.ids, expected, "exactly the questions, once each, in the mind's order"); + assert_eq!(LIMIT_MAX, 200); + assert_eq!(members.pages, 3, "599 members at the mind's page of 200"); + } + + /// The walk keeps the caller's `cites` hop: among follow-ups, only the one + /// citing Q1 is a member, and a ranking over all three counts only it. + #[test] + fn a_cites_hop_limits_the_members_to_the_cited_rows() { + let mut mind = seeded(); + committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A"), question("Q2", &["A", "B"], "A")])); + let (q1, q2) = (id("question", "Q1"), id("question", "Q2")); + committed(admit(&mut mind, vec![ + follow_up("FU-1", r(K::Question, &q1)), + follow_up("FU-2", r(K::Question, &q2)), + follow_up("FU-3", r(K::Question, &q2)), + ])); + let selection = Selection { cites: Some(cites(K::Question, &q1, None)), ..of_kinds(&[K::FollowUp]) }; + let fu = |label: &str| id("follow_up", label); + let members = mind.members(&selection, &asking(5)).unwrap(); + assert_eq!(members.ids, vec![r(K::FollowUp, &fu("FU-1"))]); + let hits = [hit(K::FollowUp, &fu("FU-2"), 0.9), hit(K::FollowUp, &fu("FU-1"), 0.8), hit(K::FollowUp, &fu("FU-3"), 0.7)]; + let ranked = rank_all(&mind, &selection, &asking(5), &hits); + assert_eq!(ranked.page.matched, 1); + assert_eq!(ids(&ranked.page), vec![fu("FU-1")]); + } + + /// The walk keeps the role of the caller's `cites` hop: spec r2 is cited + /// by two resolutions, one as its subject and one as what supersedes r1, + /// and a hop through `subject` admits only the first. + #[test] + fn a_cites_hop_keeps_its_role_in_the_members() { + let mut mind = seeded(); + let (r1, r2) = (id("cut_spec", "cut-1.r1"), id("cut_spec", "cut-1.r2")); + committed(admit(&mut mind, vec![D::CutSpec(cut_spec("1", 1))])); + committed(admit(&mut mind, vec![ + D::CutSpec(cut_spec("1", 2)), + resolution(r(K::CutSpec, &r1), superseded(&[r(K::CutSpec, &r2)])), + ])); + committed(admit(&mut mind, vec![resolution(r(K::CutSpec, &r2), withdrawn())])); + let (as_superseder, as_subject) = (id("resolution", "cut_spec.cut-1.r1.n1"), id("resolution", "cut_spec.cut-1.r2.n1")); + let through = |role: Option<&str>| Selection { + cites: Some(cites(K::CutSpec, &r2, role)), + ..of_kinds(&[K::Resolution]) + }; + let hits = [hit(K::Resolution, &as_superseder, 0.9), hit(K::Resolution, &as_subject, 0.8)]; + + let subject = through(Some("subject")); + assert_eq!(mind.members(&subject, &asking(5)).unwrap().ids, vec![r(K::Resolution, &as_subject)]); + let ranked = rank_all(&mind, &subject, &asking(5), &hits); + assert_eq!(ranked.page.matched, 1); + assert_eq!(ids(&ranked.page), vec![as_subject.clone()]); + + let superseded_by = through(Some("superseded_by")); + assert_eq!(mind.members(&superseded_by, &asking(5)).unwrap().ids, vec![r(K::Resolution, &as_superseder)]); + + let any = mind.members(&through(None), &asking(5)).unwrap(); + assert_eq!(any.ids, vec![r(K::Resolution, &as_superseder), r(K::Resolution, &as_subject)]); + } + + /// `matched` is what the selection admits, not what the index returned: + /// two hits among five members still say five. + #[test] + fn matched_counts_the_members_not_the_hits() { + let mut mind = seeded(); + committed(admit(&mut mind, (0..5).map(question_n).collect())); + let key = |n: u32| id("question", &format!("Q{n}")); + let selection = of_kinds(&[K::Question]); + let ranked = rank_all(&mind, &selection, &asking(10), &[hit(K::Question, &key(3), 0.9), hit(K::Question, &key(1), 0.2)]); + assert_eq!(ranked.page.matched, 5); + assert_eq!(ids(&ranked.page), vec![key(3), key(1)]); + } + + /// A semantic page says how near each row is, row for row and best first; + /// a plain page says nothing of the kind. + #[test] + fn a_semantic_page_carries_one_score_per_row_best_first_and_a_plain_page_none() { + let shelf = shelf(); + let hits = [ + hit(K::Question, &shelf.q2, 0.4), + hit(K::Ruling, &shelf.r4, 0.5), + hit(K::Question, &shelf.q1, 0.9), + hit(K::Question, &id("question", "Q99"), 0.8), + ]; + let page = rank_all(&shelf.mind, &standing(Selection::default()), &asking(10), &hits).page; + assert_eq!(ids(&page), vec![shelf.r4.clone(), shelf.q2.clone()]); + assert_eq!(page.scores, Some(vec![0.5, 0.4]), "the scores of the rows kept, in row order"); + let all = rank_all(&shelf.mind, &Selection::default(), &asking(2), &hits).page; + assert_eq!(ids(&all), vec![shelf.q1.clone(), shelf.r4.clone()]); + assert_eq!(all.scores, Some(vec![0.9, 0.5]), "cut to top_k, the scores are cut with the rows"); + let empty = rank_all(&shelf.mind, &standing(Selection::default()), &asking(10), &[]).page; + assert_eq!(empty.scores, Some(vec![]), "a semantic answer with no rows still says it is semantic"); + assert_eq!(shelf.mind.query(&standing(Selection::default())).unwrap().scores, None); + } + + /// The answer is the snapshot the members were taken at, not the head: a + /// ruling superseded after `members` is still in force there, and a hit + /// the members do not hold is dropped and reported. + #[test] + fn rank_reads_the_snapshot_the_members_were_taken_at() { + let mut shelf = shelf(); + let selection = standing(of_kinds(&[K::Ruling])); + let members = shelf.mind.members(&selection, &asking(10)).unwrap(); + assert!(members.ids.contains(&r(K::Ruling, &shelf.r4))); + let r5 = id("ruling", "R5"); + committed(admit(&mut shelf.mind, vec![D::Ruling(ruling("R5"))])); + committed(admit(&mut shelf.mind, vec![resolution(r(K::Ruling, &shelf.r4), superseded(&[r(K::Ruling, &r5)]))])); + assert!( + !ids(&shelf.mind.query(&selection).unwrap()).contains(&shelf.r4), + "at the head R4 no longer stands" + ); + let hits = [hit(K::Ruling, &shelf.r4, 0.5), hit(K::Ruling, &r5, 0.9)]; + let ranked = shelf.mind.rank(&selection, &asking(10), &members, &hits).expect("the members rank"); + assert_eq!(ids(&ranked.page), vec![shelf.r4.clone()], "R4 still stands as of the members' snapshot"); + assert_eq!(ranked.page.as_of, members.as_of); + assert!(members.as_of < receipt::head(&shelf.mind).unwrap()); + assert_eq!(ranked.dropped, vec![r(K::Ruling, &r5)], "a hit outside the members is dropped, not answered"); } /// A summary is bounded by a named constant whatever the document's lists diff --git a/crates/huginn-mind/src/wire.rs b/crates/huginn-mind/src/wire.rs index e0c198d..ab079c7 100644 --- a/crates/huginn-mind/src/wire.rs +++ b/crates/huginn-mind/src/wire.rs @@ -119,7 +119,7 @@ impl HuginnMindRequest { /// Every answer a mind gives. A refusal is data in the response, never a /// transport failure: the read side carries `Refused`, the write side carries /// the outcome's own `Refused`. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)] pub enum HuginnMindResponse { Whoami(MindStatus), Admit(PipelineAdmissionOutcome), @@ -263,6 +263,7 @@ mod tests { } fn responses() -> Vec { + let one_row = seeded().query(&Selection { limit: Some(1), ..Selection::default() }).unwrap().items; let refusal = MindRefusal::ForeignInstance { declared: OTHER_INSTANCE.into(), mind: INSTANCE.into() }; vec![ HuginnMindResponse::Whoami(seeded().status(IndexStatus::Current)), @@ -276,6 +277,15 @@ mod tests { next: None, items: PipelinePageItems::Headers(vec![]), edges: None, + scores: None, + }), + HuginnMindResponse::Query(PipelineSelectionPage { + matched: 1, + as_of: 1, + next: None, + items: one_row, + edges: None, + scores: Some(vec![0.5]), }), HuginnMindResponse::Refused(refusal), HuginnMindResponse::Persona(None), @@ -326,6 +336,24 @@ mod tests { } } + /// A new client reads an old daemon's page: the five fields it had, no + /// `scores`. This is what keeps rollback by flip safe. + #[test] + fn a_page_encoded_without_scores_decodes_with_none() { + #[derive(Serialize)] + struct OldPage { + matched: u32, + as_of: u64, + next: Option, + items: PipelinePageItems, + edges: Option>, + } + let old = OldPage { matched: 0, as_of: 7, next: None, items: PipelinePageItems::Headers(vec![]), edges: None }; + let bytes = rmp_serde::to_vec_named(&old).unwrap(); + let page = rmp_serde::from_slice::(&bytes).unwrap(); + assert_eq!((page.matched, page.as_of, page.scores), (0, 7, None)); + } + /// The two files the catalog publishes are the derivation of the two /// types, byte for byte. #[test] diff --git a/schemas/cultnet/huginn.mind_request.v1.schema.json b/schemas/cultnet/huginn.mind_request.v1.schema.json index 77ee834..0059e46 100644 --- a/schemas/cultnet/huginn.mind_request.v1.schema.json +++ b/schemas/cultnet/huginn.mind_request.v1.schema.json @@ -1857,7 +1857,7 @@ ] }, "SemanticQuery": { - "description": "A semantic query: the text to find documents near, and how many. The index\nowns which fields of each kind are text (`index_text`), so no substring\nfilter lives here to become a second answer to that question.", + "description": "A semantic query: the text to find documents near, and how many. The index\nowns which fields of each kind are text (`index_text`), so no substring\nfilter lives here to become a second answer to that question. `top_k` is\nthe most rows answered, all of them members of the selection: the\nselection decides membership before the index is asked.", "type": "object", "properties": { "text": { diff --git a/schemas/cultnet/huginn.mind_response.v1.schema.json b/schemas/cultnet/huginn.mind_response.v1.schema.json index 28d1fa2..0303dfc 100644 --- a/schemas/cultnet/huginn.mind_response.v1.schema.json +++ b/schemas/cultnet/huginn.mind_response.v1.schema.json @@ -3663,7 +3663,7 @@ ] }, "PipelineSelectionPage": { - "description": "One page of a selection. `matched` is the count before paging, `as_of` the\nsnapshot the page and every cursor of its walk are exact as of, and `edges`\nis present exactly when the selection follows a hop.", + "description": "One page of a selection. `matched` is the count before paging, `as_of` the\nsnapshot the page and every cursor of its walk are exact as of, `edges`\nis present exactly when the selection follows a hop, and `scores` exactly\nwhen the answer is semantic: one cosine similarity per item, in order\n(higher is nearer), for display and thresholding. It is never stored.", "type": "object", "properties": { "as_of": { @@ -3693,6 +3693,16 @@ "string", "null" ] + }, + "scores": { + "type": [ + "array", + "null" + ], + "items": { + "type": "number", + "format": "float" + } } }, "required": [