GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge eureka-substrate/semantic-members-scores: a semantic query returns up to top_k members of its selection, each with its score (Soul s4 holds)

Browse source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-25 22:39:15 +01:00
commit dd07485b4e
15 changed files with 618 additions and 174 deletions
Download patch file Download diff file Expand all files Collapse all files

2
crates/eureka-state/src/main.rs
View file

@ -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<QueryInput>) -> CallToolResult {
self.answer(move |instance| HuginnMindRequest::Query {

3
crates/eureka-state/tests/cli.rs
View file

@ -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<PipelineDocumentView>) -> Server {
next: None,
items: PipelinePageItems::Documents(stored.clone()),
edges: None,
scores: None,
};
answer("query", &HuginnMindResponse::Query(page))
}

2
crates/eureka-state/tests/common/mod.rs
View file

@ -89,7 +89,7 @@ impl<S: MindStore> IndexSink<S> for Inert {
fn status(&self) -> IndexStatus {
IndexStatus::Current
}
fn search(&mut self, _text: &str, _top_k: u32) -> Result<SearchTicket, String> {
fn search(&mut self, _text: &str, _top_k: u32, _among: Vec<String>) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn search_deadline(&self) -> Duration {

48
crates/huginn-daemon/src/daemon.rs
View file

@ -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<S: MindStore> {
fn committed(&mut self, mind: &Mind<S>, 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<SearchTicket, String>;
/// 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<String>) -> Result<SearchTicket, String>;
/// 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<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
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<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
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<Hits, String>) -> 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<SearchTicket, String> {
fn search(&mut self, _text: &str, _top_k: u32, _among: Vec<String>) -> Result<SearchTicket, String> {
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<SearchTicket, String> {
fn search(&mut self, _text: &str, _top_k: u32, _among: Vec<String>) -> Result<SearchTicket, String> {
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<SearchTicket, String> {
fn search(&mut self, _text: &str, _top_k: u32, _among: Vec<String>) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}

10
crates/huginn-daemon/src/index/fakes.rs
View file

@ -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<Hit>,
/// 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<Vec<String>>,
/// 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<Vec<Hit>> {
fn search(&mut self, collection: &str, vector: &[f32], among: &[String], limit: u32, within: Duration) -> Result<Vec<Hit>> {
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())
}
}

39
crates/huginn-daemon/src/index/mod.rs
View file

@ -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<BTreeSet<String>>;
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<Vec<Hit>>;
/// 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<Vec<Hit>>;
}
/// The mind's collection: one per instance.
@ -556,11 +554,12 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
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<E: Embedder, V: VectorIndex> Projector<E, V> {
/// 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<Hits> {
pub fn search(&mut self, text: &str, top_k: u32, among: &[String], within: Duration) -> Result<Hits> {
if let Some(why) = unsearchable(&self.progress_status()) {
bail!("{why}");
}
@ -592,8 +591,8 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
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<String> = 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<String>,
due: Instant,
}
@ -780,9 +781,9 @@ fn serve_search<E: Embedder, V: VectorIndex>(projector: &mut Projector<E, V>, 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<S: MindStore> IndexSink<S> 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<SearchTicket, String> {
fn search(&mut self, text: &str, top_k: u32, among: Vec<String>) -> Result<SearchTicket, String> {
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<S: MindStore> IndexSink<S> 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)

11
crates/huginn-daemon/src/index/qdrant.rs
View file

@ -138,9 +138,16 @@ impl VectorIndex for QdrantIndex {
Ok(())
}
fn search(&mut self, collection: &str, vector: &[f32], limit: u32, within: Duration) -> Result<Vec<Hit>> {
fn search(&mut self, collection: &str, vector: &[f32], among: &[String], limit: u32, within: Duration) -> Result<Vec<Hit>> {
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")?;

97
crates/huginn-daemon/src/index/tests.rs
View file

@ -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<String> = 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::<Vec<_>>(), [
(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::<Vec<_>>(), [
(collection().as_str(), 4, 3),
(collection().as_str(), 4, 100),
], "top_k as asked, with no oversample");
let points: Vec<String> = 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<Hits, String>)>;
/// 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<SearchTicket, String> {
<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::search(sink, text, top_k)
<WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::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 = <WorkerSink as IndexSink<OwnedRedbMessagePackBackingStore>>::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<FakeEmbedder, FakeIndex>, 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();

95
crates/huginn-daemon/src/semantic_tests.rs
View file

@ -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<Hit> = (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::<Vec<_>>(), "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::<Vec<_>>());
let all = ask(&mut client, "m-2", &semantic(rulings, 200));
assert_eq!(ids(&all), (0..20).map(kept).collect::<Vec<_>>());
let limits: Vec<u32> = 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

19
crates/huginn-daemon/tests/live_index_smoke.rs
View file

@ -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<String> = 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<String> = 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());
}
}));

2
crates/huginn-mind/src/lib.rs
View file

@ -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};

420
crates/huginn-mind/src/query.rs
View file

@ -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<String>,
pub items: PipelinePageItems,
pub edges: Option<Vec<PipelineEdge>>,
pub scores: Option<Vec<f32>>,
}
/// 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<PipelineRef>,
/// 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<S: MindStore> Mind<S> {
};
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<S: MindStore> Mind<S> {
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<Members, MindRefusal> {
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<Ranked, MindRefusal> {
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<S: MindStore> Mind<S> {
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<f32> = 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<SelectionRow>)> = std::mem::take(&mut evaluation.edges)
@ -589,7 +639,8 @@ impl<S: MindStore> Mind<S> {
.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<S: MindStore> Mind<S> {
selection: &Selection,
evaluation: Evaluation<SelectionRow>,
as_of: u64,
scores: Option<Vec<f32>>,
) -> Result<PipelineSelectionPage, MindRefusal> {
let items = if selection.projection == PROJECTION_DOCUMENT {
PipelinePageItems::Documents(evaluation.rows.iter().map(|row| row.view.clone()).collect())
@ -616,7 +668,7 @@ impl<S: MindStore> Mind<S> {
} 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<MemoryStore>, 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::<String>::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::<String>::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<String> = (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<MemoryStore>, 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<MemoryStore>, selection: &Selection) -> Vec<PipelineRef> {
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<D> =
(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<u32> = (0..599).collect();
numbered.sort_by_key(|n| (n / crate::admission::BATCH_MAX as u32, format!("Q{n}")));
let expected: Vec<PipelineRef> = 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

30
crates/huginn-mind/src/wire.rs
View file

@ -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<HuginnMindResponse> {
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<String>,
items: PipelinePageItems,
edges: Option<Vec<crate::query::PipelineEdge>>,
}
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::<PipelineSelectionPage>(&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]

2
schemas/cultnet/huginn.mind_request.v1.schema.json
View file

@ -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": {

12
schemas/cultnet/huginn.mind_response.v1.schema.json
View file

@ -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": [