Cut 11b: the semantic read, joined back through the mind

Qdrant supplies candidate ids and scores and nothing else. Mind::rank drops
every hit the mind does not hold, then lets CultNet's selection decide
membership over the whole mind with its keys narrowed to the candidates, so
in_force and cited mean what they mean in query. Order is score then ordinal;
the page is cut to the selection's limit and top_k; next is never set. A
cursor with semantic, blank text and an out-of-range top_k are refused by name
before anything is embedded. Mind::query loses its semantic parameter and the
"index is not wired" refusal; the published request schema's description of
SemanticQuery follows.

The serve loop never waits on the embedder: a semantic query becomes a Search
held beside its session, the worker embeds (behind the model's query
instruction) and searches between batches, and the loop answers when the
worker has or when ServeOptions::search_timeout passes. A worker that is
Failing or Refused refuses at once, typed Unavailable naming its own error.
VectorIndex gains search; Qdrant's adapter sends no filter.

Model identity is re-read before every flush and before every search: a model
re-pulled under the same dimensions but a new digest forgets the
reconciliation instead of mixing vectors, and the next step rebuilds. An idle
worker forgets the reconciliation every Backoff::recheck (60 s), so a lost
collection is found without a write. Both go through Projector::reconcile,
still the one judge.

Also fixes a race in the worker loop found by the mutation runs: entries were
taken from the inbox under one lock and published as wanted under another, so
whoami could read a stale Current with nothing queued while the worker held
them. Taking, absorbing and publishing are now one critical section.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

README.md

@@ -24,8 +24,12 @@ Upstream: `https://github.com/GameCult/Huginn.git`
   restart compares the mind with the collection and writes what is missing. An
   unreachable Qdrant or Ollama shows in `whoami` as the index's status and is
   retried; admission never waits on it. The daemon requires `--qdrant-url`,
-  `--ollama-url` and `--embedding-model`. The semantic read side is not wired
-  yet.
+  `--ollama-url` and `--embedding-model`. A `query` carrying `semantic` is
+  answered through the same selection as any other: the index supplies
+  candidate ids and scores, and the mind keeps only the documents it holds and
+  the selection admits (ask for `in_force` to exclude resolved ones), ordered by
+  score. It takes no cursor, and it is refused `Unavailable` while the index is
+  failing or refused.
 - Mind state is not version-controlled. It lives in Huginn's store, not in
   any repository.
 

crates/huginn-daemon/src/daemon.rs

@@ -3,18 +3,67 @@
 //! thing that reads a wall clock.
 
 use chrono::{DateTime, Utc};
+use cultnet_rs::Selection;
 use huginn_mind::epiphany_pipeline::{PipelineRef, Slug};
 use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, IndexStatus};
-use huginn_mind::{Mind, MindStore, PipelineAdmissionOutcome};
+use huginn_mind::{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
 /// `mind.view`. Its error is logged and never reaches the caller: the index
 /// does not decide whether a batch was admitted, and the call must not wait on
 /// the index's transport. `status` is what `whoami` reports of it.
+///
+/// A semantic query is asked in two steps because the index answers on its own
+/// time: `search` starts one and returns a ticket at once, and `searched`
+/// hands over the searches that have finished since it was last called. The
+/// candidates are the index's word about nearness and nothing else; the daemon
+/// judges them through the mind.
 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>;
+    /// Every search that finished since the last call, by ticket: its
+    /// candidates and scores, or why it failed.
+    fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)>;
+}
+
+/// Names one search the index was asked, so the loop can match the answer to
+/// the client that is waiting for it.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct SearchTicket(pub u64);
+
+/// What a search returns: candidate documents and how near each scored, best
+/// first. Nothing else about them.
+pub type Hits = Vec<(PipelineRef, f32)>;
+
+/// A semantic query the index has been asked and has not yet answered. The
+/// serve loop holds it beside the session that asked, and hands it back to
+/// `Daemon::finish` with the index's answer.
+pub struct Search {
+    pub ticket: SearchTicket,
+    selection: Selection,
+    semantic: SemanticQuery,
+}
+
+/// What a request came to: an answer, or a search the index has yet to
+/// finish.
+pub enum Handled {
+    Answered(HuginnMindResponse),
+    Searching(Search),
+}
+
+impl Handled {
+    /// The answer, for a request that cannot be a search.
+    #[cfg(test)]
+    pub(crate) fn answered(self) -> HuginnMindResponse {
+        match self {
+            Self::Answered(response) => response,
+            Self::Searching(search) => panic!("expected an answer, got the search {:?}", search.ticket),
+        }
+    }
 }
 
 /// The envelope's `source_runtime_id` on every response a daemon serving this
@@ -54,20 +103,20 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
         runtime_id(self.mind.instance())
     }
 
-    /// One request, one answer. `Admit` goes to the mind whole, because the
+    /// One request, one answer, or a search to finish. `Admit` goes to the mind whole, because the
     /// batch declares its own instance and admission's A1 is the check. Every
     /// other request names an instance: `require_instance` now applies the
     /// leaf's own grammar before it compares identity, so a name outside the
     /// grammar (a fullwidth fold, say) is refused by name rather than read as
     /// merely a foreign mind, and the daemon asks nothing twice.
-    pub fn handle(&mut self, request: HuginnMindRequest, now: DateTime<Utc>) -> HuginnMindResponse {
+    pub fn handle(&mut self, request: HuginnMindRequest, now: DateTime<Utc>) -> Handled {
         if let Some(declared) = request.instance()
             && !matches!(request, HuginnMindRequest::Admit(_))
             && let Err(refusal) = self.mind.require_instance(declared)
         {
-            return HuginnMindResponse::Refused(refusal);
+            return Handled::Answered(HuginnMindResponse::Refused(refusal));
         }
-        match request {
+        let answer = match request {
             HuginnMindRequest::Whoami => HuginnMindResponse::Whoami(self.mind.status(self.index.status())),
             HuginnMindRequest::Admit(batch) => {
                 let outcome = self.mind.admit(batch, now);
@@ -83,10 +132,57 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
                 Ok(view) => HuginnMindResponse::View(view),
                 Err(refusal) => HuginnMindResponse::Refused(refusal),
             },
-            HuginnMindRequest::Query { selection, semantic, .. } => match self.mind.query(&selection, semantic.as_ref()) {
+            HuginnMindRequest::Query { selection, semantic: None, .. } => match self.mind.query(&selection) {
                 Ok(page) => HuginnMindResponse::Query(page),
                 Err(refusal) => HuginnMindResponse::Refused(refusal),
             },
+            HuginnMindRequest::Query { selection, semantic: Some(semantic), .. } => {
+                return self.begin_search(selection, semantic);
+            }
+        };
+        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.
+    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);
+        }
+        match self.index.search(&semantic.text.0, semantic.top_k) {
+            Ok(ticket) => Handled::Searching(Search { ticket, selection, semantic }),
+            Err(detail) => refused(MindRefusal::Unavailable { detail }),
+        }
+    }
+
+    /// Searches the index has finished since this was last called.
+    pub fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
+        self.index.searched()
+    }
+
+    /// The ids among `hits` that the mind holds no document for: the index
+    /// keeps points the mind no longer has (a restored mind, a second store),
+    /// and a hit is never shown on the index's word alone.
+    pub(crate) fn unheld<'a>(&self, hits: &'a Hits) -> Vec<&'a str> {
+        hits.iter().filter(|(id, _)| !matches!(self.mind.view(id), Ok(Some(_)))).map(|(id, _)| id.id.0.as_str()).collect()
+    }
+
+    /// 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
+    /// `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 }),
+        };
+        for id in self.unheld(&hits) {
+            eprintln!("huginn: the index returned {id}, which this mind does not hold; it is left out");
+        }
+        match self.mind.rank(&search.selection, &search.semantic, &hits) {
+            Ok(page) => HuginnMindResponse::Query(page),
+            Err(refusal) => HuginnMindResponse::Refused(refusal),
         }
     }
 }
@@ -105,7 +201,7 @@ pub(crate) mod tests {
     use huginn_mind::{MindRefusal, OwnedRedbMessagePackBackingStore};
     use std::path::Path;
     use cultnet_rs::Selection;
-    use huginn_mind::{Faculty, PipelineAdmissionBatch, PipelineProvenance, PipelinePageItems, PipelineStatus, SemanticQuery};
+    use huginn_mind::{Faculty, PipelineAdmissionBatch, PipelineProvenance, PipelinePageItems, PipelineStatus};
     use std::sync::{Arc, Mutex};
     use tempfile::TempDir;
 
@@ -300,7 +396,7 @@ pub(crate) mod tests {
             campaign_seed(),
             vec![cut_spec(WIDE_CUT, WIDE_CHANGES), cut_spec(FITTING_CUT, FITTING_CHANGES)],
         ] {
-            let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), now());
+            let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), now()).answered();
             assert!(
                 matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })),
                 "{outcome:?}"
@@ -315,7 +411,7 @@ pub(crate) mod tests {
     pub(crate) fn seeded() -> (TempDir, Daemon<OwnedRedbMessagePackBackingStore, NoIndex>) {
         let root = tempfile::tempdir().unwrap();
         let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
-        let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
+        let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()).answered();
         assert!(
             matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })),
             "{outcome:?}"
@@ -334,6 +430,14 @@ pub(crate) mod tests {
         fn status(&self) -> IndexStatus {
             IndexStatus::Current
         }
+
+        fn search(&mut self, _text: &str, _top_k: u32) -> Result<SearchTicket, String> {
+            Err("this index cannot search".into())
+        }
+
+        fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
+            Vec::new()
+        }
     }
 
     /// A daemon over the instance's mind under `state_root`, with no index.
@@ -360,6 +464,14 @@ pub(crate) mod tests {
         fn status(&self) -> IndexStatus {
             IndexStatus::Current
         }
+
+        fn search(&mut self, _text: &str, _top_k: u32) -> Result<SearchTicket, String> {
+            Err("this index cannot search".into())
+        }
+
+        fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
+            Vec::new()
+        }
     }
 
     struct FailingIndex;
@@ -372,10 +484,18 @@ pub(crate) mod tests {
         fn status(&self) -> IndexStatus {
             IndexStatus::Current
         }
+
+        fn search(&mut self, _text: &str, _top_k: u32) -> Result<SearchTicket, String> {
+            Err("this index cannot search".into())
+        }
+
+        fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
+            Vec::new()
+        }
     }
 
     fn status(daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, impl IndexSink<OwnedRedbMessagePackBackingStore>>) -> MindStatus {
-        match daemon.handle(HuginnMindRequest::Whoami, now()) {
+        match daemon.handle(HuginnMindRequest::Whoami, now()).answered() {
             HuginnMindResponse::Whoami(status) => status,
             other => panic!("expected a status, got {other:?}"),
         }
@@ -387,7 +507,7 @@ pub(crate) mod tests {
     fn a_batch_round_trips_typed_through_handle_without_a_socket() {
         let root = tempfile::tempdir().unwrap();
         let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
-        let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
+        let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()).answered();
         let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = outcome else {
             panic!("expected a commit");
         };
@@ -410,7 +530,7 @@ pub(crate) mod tests {
         let HuginnMindResponse::Query(page) = daemon.handle(
             HuginnMindRequest::Query { instance: slug(INSTANCE), selection: instances.clone(), semantic: None },
             now(),
-        ) else {
+        ).answered() else {
             panic!("expected a page");
         };
         assert_eq!(page.matched, 1);
@@ -421,13 +541,13 @@ pub(crate) mod tests {
         let id = writes[0].clone();
         let whole = Selection { projection: "document".into(), ..instances };
         let HuginnMindResponse::Query(page) =
-            daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection: whole, semantic: None }, now())
+            daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection: whole, semantic: None }, now()).answered()
         else {
             panic!("expected a page");
         };
         let PipelinePageItems::Documents(views) = &page.items else { panic!("the document projection is the view") };
         assert_eq!(views[0].status, PipelineStatus::InForce);
-        let view = daemon.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id: id.clone() }, now());
+        let view = daemon.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id: id.clone() }, now()).answered();
         assert_eq!(view, HuginnMindResponse::View(Some(views[0].clone())));
     }
 
@@ -437,18 +557,18 @@ pub(crate) mod tests {
     fn a_read_or_a_write_naming_another_instance_is_refused_by_the_mind_and_writes_nothing() {
         let root = tempfile::tempdir().unwrap();
         let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
-        let landed = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
+        let landed = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()).answered();
         let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = landed else {
             panic!("expected a commit");
         };
         let foreign = MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() };
 
         let query = HuginnMindRequest::Query { instance: slug(OTHER), selection: Selection::default(), semantic: None };
-        assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(foreign.clone()));
+        assert_eq!(daemon.handle(query, now()).answered(), HuginnMindResponse::Refused(foreign.clone()));
 
         let admit = HuginnMindRequest::Admit(batch(OTHER, vec![identity(OTHER)]));
         assert_eq!(
-            daemon.handle(admit, now()),
+            daemon.handle(admit, now()).answered(),
             HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(foreign.clone()))
         );
 
@@ -459,12 +579,12 @@ pub(crate) mod tests {
         let (question, _) = question_and_ruling();
         let admit = HuginnMindRequest::Admit(batch(OTHER, vec![question]));
         assert_eq!(
-            daemon.handle(admit, now()),
+            daemon.handle(admit, now()).answered(),
             HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(foreign.clone()))
         );
 
         let view = HuginnMindRequest::View { instance: slug(OTHER), id: writes[0].clone() };
-        assert_eq!(daemon.handle(view, now()), HuginnMindResponse::Refused(foreign));
+        assert_eq!(daemon.handle(view, now()).answered(), HuginnMindResponse::Refused(foreign));
 
         // Three names a comparison could mistake for this mind's. `NEAR` is
         // `INSTANCE`'s own length and differs in its last byte, so a check
@@ -479,12 +599,12 @@ pub(crate) mod tests {
         for declared in [NEAR, PREFIXED, CASED] {
             let refusal = MindRefusal::ForeignInstance { declared: declared.into(), mind: INSTANCE.into() };
             let query = HuginnMindRequest::Query { instance: slug(declared), selection: Selection::default(), semantic: None };
-            assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(refusal.clone()), "{declared}");
+            assert_eq!(daemon.handle(query, now()).answered(), HuginnMindResponse::Refused(refusal.clone()), "{declared}");
             let view = HuginnMindRequest::View { instance: slug(declared), id: writes[0].clone() };
-            assert_eq!(daemon.handle(view, now()), HuginnMindResponse::Refused(refusal.clone()), "{declared}");
+            assert_eq!(daemon.handle(view, now()).answered(), HuginnMindResponse::Refused(refusal.clone()), "{declared}");
             let admit = HuginnMindRequest::Admit(batch(declared, vec![identity(declared)]));
             assert_eq!(
-                daemon.handle(admit, now()),
+                daemon.handle(admit, now()).answered(),
                 HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(refusal)),
                 "{declared}"
             );
@@ -514,13 +634,13 @@ pub(crate) mod tests {
         });
 
         let query = HuginnMindRequest::Query { instance: slug(fullwidth), selection: Selection::default(), semantic: None };
-        assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(expected.clone()));
+        assert_eq!(daemon.handle(query, now()).answered(), HuginnMindResponse::Refused(expected.clone()));
 
         let view = HuginnMindRequest::View {
             instance: slug(fullwidth),
             id: PipelineRef { kind: PipelineKind::Question, id: Short("not-reached".into()) },
         };
-        assert_eq!(daemon.handle(view, now()), HuginnMindResponse::Refused(expected));
+        assert_eq!(daemon.handle(view, now()).answered(), HuginnMindResponse::Refused(expected));
     }
 
     /// `Admit` is excluded from the daemon's own pre-check (its batch is not a
@@ -545,12 +665,12 @@ pub(crate) mod tests {
 
         let admit = HuginnMindRequest::Admit(batch(fullwidth, vec![identity(INSTANCE)]));
         assert_eq!(
-            daemon.handle(admit, now()),
+            daemon.handle(admit, now()).answered(),
             HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(expected.clone()))
         );
 
         let query = HuginnMindRequest::Query { instance: slug(fullwidth), selection: Selection::default(), semantic: None };
-        assert_eq!(daemon.handle(query, now()), HuginnMindResponse::Refused(expected));
+        assert_eq!(daemon.handle(query, now()).answered(), HuginnMindResponse::Refused(expected));
     }
 
     /// A refusal a read method raised is the answer the dispatch returns, whole.
@@ -572,17 +692,7 @@ pub(crate) mod tests {
             value: "not a reference".into(),
         });
         let view = HuginnMindRequest::View { instance: slug(INSTANCE), id: invalid };
-        assert_eq!(daemon.handle(view, now()), HuginnMindResponse::Refused(malformed));
-
-        // A semantic query is unavailable until Cut 11 wires an index, and the
-        // detail is the mind's own sentence.
-        let semantic = Some(SemanticQuery { text: "who owns the state".into(), top_k: 4 });
-        let unwired =
-            MindRefusal::Unavailable { detail: "semantic query: the index is not wired (Cut 11)".into() };
-        assert_eq!(
-            daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection: Selection::default(), semantic }, now()),
-            HuginnMindResponse::Refused(unwired)
-        );
+        assert_eq!(daemon.handle(view, now()).answered(), HuginnMindResponse::Refused(malformed));
 
         // And the same reads answer rather than refuse when they are asked
         // something answerable, so the arms above are refusing on the mind's
@@ -590,9 +700,9 @@ pub(crate) mod tests {
         let absent =
             PipelineRef { kind: PipelineKind::Question, id: Short(format!("{CAMPAIGN}:question:Q1")) };
         let present = HuginnMindRequest::View { instance: slug(INSTANCE), id: absent };
-        assert_eq!(daemon.handle(present, now()), HuginnMindResponse::View(None));
+        assert_eq!(daemon.handle(present, now()).answered(), HuginnMindResponse::View(None));
         let plain = HuginnMindRequest::Query { instance: slug(INSTANCE), selection: Selection::default(), semantic: None };
-        let HuginnMindResponse::Query(page) = daemon.handle(plain, now()) else { panic!("expected a page") };
+        let HuginnMindResponse::Query(page) = daemon.handle(plain, now()).answered() else { panic!("expected a page") };
         assert_eq!(page.matched, 1);
     }
 
@@ -604,7 +714,7 @@ pub(crate) mod tests {
     fn a_selection_refusal_crosses_the_dispatch_as_itself() {
         let (_root, mut daemon) = seeded();
         let ask = |daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, NoIndex>, selection: Selection| {
-            daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection, semantic: None }, now())
+            daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection, semantic: None }, now()).answered()
         };
         let predicate = |index: &str, value: &str| Selection {
             fields: Some(vec![cultnet_rs::FieldPredicate {
@@ -641,7 +751,7 @@ pub(crate) mod tests {
         let mut daemon = daemon.with_index(recording.clone());
         let (question, ruling) = question_and_ruling();
 
-        let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now());
+        let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now()).answered();
         let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = outcome else {
             panic!("expected a commit");
         };
@@ -657,13 +767,13 @@ pub(crate) mod tests {
         let mut daemon = daemon.with_index(FailingIndex);
         let root = tempfile::tempdir().unwrap();
         let mut fresh = open_unindexed(root.path(), &slug(INSTANCE)).unwrap().with_index(FailingIndex);
-        fresh.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
-        let outcome = fresh.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now());
+        fresh.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()).answered();
+        let outcome = fresh.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now()).answered();
         let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = outcome else {
             panic!("a failing index does not refuse the batch");
         };
         for write in &writes {
-            let view = fresh.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id: write.clone() }, now());
+            let view = fresh.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id: write.clone() }, now()).answered();
             assert!(matches!(view, HuginnMindResponse::View(Some(_))), "{write:?} is readable");
         }
         assert_eq!(status(&mut daemon).documents, 4, "identity, question, ruling and the derived resolution");
@@ -684,7 +794,7 @@ pub(crate) mod tests {
         drop(held);
 
         let mut planted = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
-        planted.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
+        planted.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now()).answered();
         drop(planted);
         let moved = Mind::<OwnedRedbMessagePackBackingStore>::store_path_for(root.path(), &slug(OTHER)).unwrap();
         std::fs::create_dir_all(moved.parent().unwrap()).unwrap();

crates/huginn-daemon/src/index/fakes.rs

@@ -5,9 +5,9 @@
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::{Arc, Condvar, Mutex};
 
-use anyhow::{Result, bail};
+use anyhow::{Result, bail, ensure};
 
-use super::{CollectionMeta, Described, Embedder, ModelIdentity, Point, VectorIndex};
+use super::{CollectionMeta, Described, Embedder, Hit, ModelIdentity, Point, VectorIndex};
 
 pub(crate) fn identity(digest: &str) -> ModelIdentity {
     ModelIdentity { name: "model".into(), digest: digest.into(), dimensions: 4 }
@@ -30,6 +30,13 @@ impl Gate {
         }
     }
 
+    /// Shuts the door again and forgets who arrived, so a test that let start-up
+    /// work through can hold the next call.
+    pub(crate) fn shut_again(&self) {
+        let mut state = self.0.0.lock().unwrap();
+        *state = (0, false);
+    }
+
     pub(crate) fn open(&self) {
         self.0.0.lock().unwrap().1 = true;
         self.0.1.notify_all();
@@ -109,6 +116,11 @@ pub(crate) struct IndexState {
     pub(crate) recreated: Vec<(String, CollectionMeta)>,
     /// The point ids of every upsert call, in order.
     pub(crate) upserts: Vec<Vec<String>>,
+    /// What every search answers, whatever the vector: the test decides the
+    /// candidates, as a real index would decide them from vectors.
+    pub(crate) hits: Vec<Hit>,
+    /// The collection, vector length and limit of every search, in order.
+    pub(crate) searches: Vec<(String, usize, u32)>,
 }
 
 #[derive(Clone, Default)]
@@ -169,6 +181,16 @@ impl VectorIndex for FakeIndex {
         }
         Ok(())
     }
+
+    fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result<Vec<Hit>> {
+        let mut state = self.state.lock().unwrap();
+        if state.down {
+            bail!("the vector store is down");
+        }
+        state.searches.push((collection.into(), vector.len(), limit));
+        ensure!(state.collections.contains_key(collection), "no collection {collection}");
+        Ok(state.hits.clone())
+    }
 }
 
 /// A working embedder and vector store, for tests whose subject is elsewhere.

crates/huginn-daemon/src/index/mod.rs

@@ -21,7 +21,17 @@
 //! Reconciliation is not a startup event. After any failed step the worker
 //! reconciles again before it writes anything else: a collection lost while the
 //! daemon runs is recreated and refilled, and a model re-pulled with other
-//! dimensions rebuilds the collection, all without a restart. Refilling needs
+//! dimensions rebuilds the collection, all without a restart. Two checks keep
+//! that true without a failure to announce it. Before every flush the worker
+//! re-reads the model's identity (name, digest, dimensions; about 0.3 s against
+//! Ollama) and, if it is not the one the collection was reconciled under,
+//! forgets the reconciliation instead of writing: a model re-pulled under the
+//! same dimensions never mixes its vectors with the old model's. An idle
+//! worker forgets the reconciliation every `Backoff::recheck` (60 s by
+//! default), which re-reads the model, describes the collection and lists its
+//! points: a collection lost while nothing was being written is found within
+//! that interval, and until then `Current` is what the worker last verified.
+//! There is one judge either way: `reconcile`. Refilling needs
 //! the text, so the projector keeps every entry it was given (`known`) beside
 //! the ids it still owes (`wanted`): the whole indexable mind's text is held in
 //! memory for the process's life.
@@ -45,10 +55,14 @@
 //!   fail the whole batch forever on one dense document, so it is not set.
 //!   There is no chunking.
 //! - Points are never deleted. Withdrawn and superseded documents are indexed
-//!   on purpose and the payload carries no status. Cut 11b's semantic read must
-//!   therefore join every hit back through the mind: drop ids the mind does not
-//!   hold, and filter by resolution using RS-3's in-force derivation through
-//!   selection, never by anything in the payload.
+//!   on purpose and the payload carries no status. The semantic read therefore
+//!   takes only candidate ids and scores from a search: `Daemon::finish` joins
+//!   them back through `Mind::rank`, which drops ids the mind does not hold and
+//!   lets the selection decide the rest, in-force derivation included.
+//! - A search runs on the worker between batches, so it waits for the batch in
+//!   flight. The serve loop never waits: it holds the reply and answers when
+//!   the worker has (`ServeOptions::search_timeout` bounds the wait). A search
+//!   is refused at once while the worker is `Failing` or `Refused`.
 
 pub mod ollama;
 pub mod qdrant;
@@ -56,17 +70,18 @@ pub mod qdrant;
 #[cfg(test)]
 pub(crate) mod fakes;
 
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
 use std::sync::{Arc, Condvar, Mutex, MutexGuard};
 use std::time::{Duration, Instant};
 
-use anyhow::{Context, Result, ensure};
-use huginn_mind::epiphany_pipeline::{PipelineRef, Slug};
+use anyhow::{Context, Result, bail, ensure};
+use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef, Short, Slug};
 use huginn_mind::wire::IndexStatus;
 use huginn_mind::{INDEX_TEXT_VERSION, IndexEntry, Mind, MindStore};
 use sha2::{Digest, Sha256};
 
-use crate::daemon::IndexSink;
+use crate::daemon::{Hits, IndexSink, SearchTicket};
+use ollama::QUERY_INSTRUCTION;
 
 /// The `managed_by` a collection carries when this organ made it.
 pub const MANAGED_BY: &str = "huginn";
@@ -74,6 +89,9 @@ pub const MANAGED_BY: &str = "huginn";
 /// Documents embedded and written per call.
 pub const BATCH: usize = 32;
 
+/// 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.
@@ -123,6 +141,16 @@ pub struct Point {
     pub payload: PointPayload,
 }
 
+/// One candidate a vector search returned: the document it names, by the id and
+/// kind the payload carries, and how near it scored. Nothing else about the
+/// document is the index's to say.
+#[derive(Clone, Debug, PartialEq)]
+pub struct Hit {
+    pub doc_id: String,
+    pub kind: String,
+    pub score: f32,
+}
+
 pub trait Embedder {
     fn model_identity(&mut self) -> Result<ModelIdentity>;
     /// One vector per text, in order.
@@ -137,6 +165,9 @@ 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.
+    fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result<Vec<Hit>>;
 }
 
 /// The mind's collection: one per instance.
@@ -201,7 +232,9 @@ pub struct Projector<E: Embedder, V: VectorIndex> {
     known: BTreeMap<String, IndexEntry>,
     /// The ids in `known` whose points the collection may lack.
     wanted: BTreeSet<String>,
-    dimensions: Option<u32>,
+    /// The model the collection was reconciled under; `None` until it is, and
+    /// again whenever that reconciliation is forgotten.
+    reconciled: Option<ModelIdentity>,
 }
 
 impl<E: Embedder, V: VectorIndex> Projector<E, V> {
@@ -214,7 +247,7 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
             collection: collection_name(instance),
             known: BTreeMap::new(),
             wanted: BTreeSet::new(),
-            dimensions: None,
+            reconciled: None,
         }
     }
 
@@ -246,26 +279,56 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
     pub fn advance(&mut self) -> Result<Advance, StepError> {
         let stepped = self.step();
         if stepped.is_err() {
-            self.dimensions = None;
+            self.reconciled = None;
         }
         stepped
     }
 
+    /// `advance` after forgetting the reconciliation on purpose: the idle
+    /// worker's periodic look at whether the model and the collection are still
+    /// what it last verified.
+    pub fn recheck(&mut self) -> Result<Advance, StepError> {
+        self.reconciled = None;
+        self.advance()
+    }
+
+    /// Whether the mind has an identity and the collection has not been
+    /// reconciled under the model: `advance` has reconciling to do.
+    pub fn stale(&self) -> bool {
+        self.mind.is_some() && self.reconciled.is_none()
+    }
+
     fn step(&mut self) -> Result<Advance, StepError> {
         if self.mind.is_none() {
             return Ok(Advance::Idle);
         }
-        if self.dimensions.is_none() {
+        if self.reconciled.is_none() {
             self.reconcile()?;
             return Ok(Advance::Progressed);
         }
         if self.wanted.is_empty() {
             return Ok(Advance::Idle);
         }
+        if !self.model_unchanged()? {
+            return Ok(Advance::Progressed);
+        }
         self.flush_batch()?;
         Ok(Advance::Progressed)
     }
 
+    /// Re-reads the model's identity and compares it with the one the
+    /// collection was reconciled under. A different model, digest or length
+    /// forgets the reconciliation, so the next step reconciles (and rebuilds)
+    /// instead of writing or searching with vectors the collection cannot mix.
+    fn model_unchanged(&mut self) -> Result<bool, StepError> {
+        let now = self.embedder.model_identity().context("re-reading the embedding model's identity")?;
+        if self.reconciled.as_ref() == Some(&now) {
+            return Ok(true);
+        }
+        self.reconciled = None;
+        Ok(false)
+    }
+
     /// The only place a point is judged missing. Reads the model identity,
     /// makes the collection agree with it (creating it, or rebuilding it when
     /// the model, its digest, its dimensions or the text version differ),
@@ -306,7 +369,7 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
         }
         let present = self.index.ids(&self.collection).context("listing the collection's points")?;
         self.wanted = self.known.keys().filter(|document_id| !present.contains(&point_id(document_id))).cloned().collect();
-        self.dimensions = Some(identity.dimensions);
+        self.reconciled = Some(identity);
         Ok(())
     }
 
@@ -322,7 +385,7 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
     }
 
     fn flush(&mut self) -> Result<()> {
-        let dimensions = self.dimensions.context("flush before reconcile")? as usize;
+        let dimensions = self.reconciled.as_ref().context("flush before reconcile")?.dimensions as usize;
         let batch: Vec<IndexEntry> =
             self.wanted.iter().take(BATCH).filter_map(|document_id| self.known.get(document_id)).cloned().collect();
         if batch.is_empty() {
@@ -358,9 +421,40 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
         Ok(())
     }
 
+    /// 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 model that has changed
+    /// since the reconciliation refuses the search rather than compare vectors
+    /// of two models. A hit naming a kind this organ has no name for is logged
+    /// and left out.
+    pub fn search(&mut self, text: &str, top_k: u32) -> Result<Hits> {
+        let Some(dimensions) = self.reconciled.as_ref().map(|model| model.dimensions as usize) else {
+            bail!("the index has not been reconciled with the model and the collection");
+        };
+        ensure!(
+            self.model_unchanged().map_err(|error| anyhow::anyhow!("{error}"))?,
+            "the embedding model changed since the collection was built; the index is being rebuilt for it"
+        );
+        let query = format!("Instruct: {QUERY_INSTRUCTION}\nQuery: {text}");
+        let mut vectors = self.embedder.embed(&[query]).context("embedding the query")?;
+        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 found = self.index.search(&self.collection, &vector, top_k.saturating_mul(4).min(OVERSAMPLE_MAX))?;
+        let mut hits = Vec::with_capacity(found.len());
+        for hit in found {
+            match PipelineKind::ALL.into_iter().find(|kind| kind.name() == hit.kind) {
+                Some(kind) => hits.push((PipelineRef { kind: *kind, id: Short(hit.doc_id) }, hit.score)),
+                None => eprintln!("huginn: the index returned {} under the unknown kind {:?}; it is left out", hit.doc_id, hit.kind),
+            }
+        }
+        Ok(hits)
+    }
+
     /// The health after a step that did not fail.
     pub fn progress_status(&self) -> IndexStatus {
-        match (self.dimensions, self.pending()) {
+        match (&self.reconciled, self.pending()) {
             (None, 0) if self.mind.is_none() => IndexStatus::Current,
             (None, pending) => IndexStatus::Reconciling { pending },
             (Some(_), 0) => IndexStatus::Current,
@@ -379,17 +473,19 @@ impl<E: Embedder, V: VectorIndex> Projector<E, V> {
     }
 }
 
-/// The retry delay: it starts at `initial`, doubles per consecutive failure and
-/// stops at `max`.
+/// The worker's clock: the retry delay starts at `initial`, doubles per
+/// consecutive failure and stops at `max`; an idle worker verifies the model
+/// and the collection again every `recheck`.
 #[derive(Clone, Copy, Debug)]
 pub struct Backoff {
     pub initial: Duration,
     pub max: Duration,
+    pub recheck: Duration,
 }
 
 impl Default for Backoff {
     fn default() -> Self {
-        Self { initial: Duration::from_secs(30), max: Duration::from_secs(600) }
+        Self { initial: Duration::from_secs(30), max: Duration::from_secs(600), recheck: Duration::from_secs(60) }
     }
 }
 
@@ -405,6 +501,11 @@ struct Shared {
     mind: Option<String>,
     status: IndexStatus,
     closed: bool,
+    /// Searches started and not yet taken by the worker.
+    searches: VecDeque<(SearchTicket, String, u32)>,
+    /// Searches the worker finished and the serve loop has not collected.
+    found: Vec<(SearchTicket, Result<Hits, String>)>,
+    next_ticket: u64,
 }
 
 type Lock = Arc<(Mutex<Shared>, Condvar)>;
@@ -413,27 +514,36 @@ fn lock(shared: &Lock) -> MutexGuard<'_, Shared> {
     shared.0.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
 }
 
-/// With no timeout, blocks until the inbox holds something, the mind reveals
-/// its identity, or the sink is closed. With one, sleeps out the backoff whatever arrives: a failing index
-/// is retried on its schedule, not once per admission.
-fn wait_for_work(shared: &Lock, timeout: Option<Duration>) {
+/// How long the worker waits, and what ends the wait early.
+enum Wait {
+    /// Idle: until something arrives (an entry, the mind's identity, a search)
+    /// or `recheck` passes.
+    Work { recheck: Duration },
+    /// Backing off from a failure: the whole delay, whatever arrives. A failing
+    /// index is retried on its schedule, not once per admission.
+    Backoff(Duration),
+}
+
+/// Whether the wait ended because the recheck interval passed.
+fn wait_for_work(shared: &Lock, wait: Wait) -> bool {
+    let (timeout, wakes_on_work) = match wait {
+        Wait::Work { recheck } => (recheck, true),
+        Wait::Backoff(delay) => (delay, false),
+    };
+    let deadline = Instant::now() + timeout;
     let mut guard = lock(shared);
-    match timeout {
-        None => {
-            while guard.inbox.is_empty() && guard.mind.is_none() && !guard.closed {
-                guard = shared.1.wait(guard).unwrap_or_else(|poisoned| poisoned.into_inner());
-            }
+    loop {
+        if guard.closed {
+            return false;
         }
-        Some(timeout) => {
-            let deadline = Instant::now() + timeout;
-            while !guard.closed {
-                let left = deadline.saturating_duration_since(Instant::now());
-                if left.is_zero() {
-                    return;
-                }
-                guard = shared.1.wait_timeout(guard, left).unwrap_or_else(|poisoned| poisoned.into_inner()).0;
-            }
+        if wakes_on_work && (!guard.inbox.is_empty() || guard.mind.is_some() || !guard.searches.is_empty()) {
+            return false;
         }
+        let left = deadline.saturating_duration_since(Instant::now());
+        if left.is_zero() {
+            return wakes_on_work;
+        }
+        guard = shared.1.wait_timeout(guard, left).unwrap_or_else(|poisoned| poisoned.into_inner()).0;
     }
 }
 
@@ -460,37 +570,72 @@ fn including(status: &IndexStatus, queued: u32) -> IndexStatus {
     }
 }
 
+/// Runs every search the serve loop has queued, one at a time, and leaves each
+/// answer for the loop to collect. Between batches, never inside one.
+fn serve_searches<E: Embedder, V: VectorIndex>(projector: &mut Projector<E, V>, shared: &Lock) {
+    loop {
+        let Some((ticket, text, top_k)) = lock(shared).searches.pop_front() else { return };
+        let found = projector.search(&text, top_k).map_err(|error| format!("{error:#}"));
+        lock(shared).found.push((ticket, found));
+    }
+}
+
+/// Answers every queued search with why the index cannot: they were accepted
+/// while it was healthy and it has failed since.
+fn refuse_searches(shared: &Lock, why: &str) {
+    let mut guard = lock(shared);
+    while let Some((ticket, ..)) = guard.searches.pop_front() {
+        guard.found.push((ticket, Err(why.to_owned())));
+    }
+}
+
 fn work<E: Embedder, V: VectorIndex>(mut projector: Projector<E, V>, shared: Lock, backoff: Backoff) {
     let mut failures = 0_u32;
     let mut delay = backoff.initial;
+    let mut recheck = false;
     loop {
-        let (mind, inbox) = {
+        // The inbox is taken, absorbed and its effect published under one
+        // lock: between taking entries out of the inbox and publishing them
+        // as wanted, `status` would otherwise read a stale `Current` with
+        // nothing queued, and a reader would see the mind caught up while
+        // entries were in the worker's hands.
+        {
             let mut guard = lock(&shared);
             if guard.closed {
                 return;
             }
-            (guard.mind.take(), std::mem::take(&mut guard.inbox))
-        };
-        if let Some(mind) = mind {
-            projector.identify(mind);
+            let mind = guard.mind.take();
+            let inbox = std::mem::take(&mut guard.inbox);
+            if let Some(mind) = mind {
+                projector.identify(mind);
+            }
+            projector.want(inbox);
+            // While a failure is being reported, the report stands until an
+            // attempt succeeds: publishing progress between retries would
+            // flicker.
+            if failures == 0 {
+                guard.status = projector.progress_status();
+            }
         }
-        projector.want(inbox);
-        // While a failure is being reported, the report stands until an
-        // attempt succeeds: publishing progress between retries would flicker.
-        if failures == 0 {
-            publish(&shared, projector.progress_status());
-        }
-        match projector.advance() {
+        shared.1.notify_all();
+        let stepped = if std::mem::take(&mut recheck) { projector.recheck() } else { projector.advance() };
+        match stepped {
             Ok(Advance::Progressed) => {
                 failures = 0;
                 delay = backoff.initial;
                 publish(&shared, projector.progress_status());
+                serve_searches(&mut projector, &shared);
             }
             Ok(Advance::Idle) => {
                 failures = 0;
                 delay = backoff.initial;
                 publish(&shared, projector.progress_status());
-                wait_for_work(&shared, None);
+                serve_searches(&mut projector, &shared);
+                // A search that found the model changed has left the
+                // reconciliation to redo: do it now, not at the next wake.
+                if !projector.stale() {
+                    recheck = wait_for_work(&shared, Wait::Work { recheck: backoff.recheck });
+                }
             }
             Err(error) => {
                 failures = failures.saturating_add(1);
@@ -500,7 +645,8 @@ fn work<E: Embedder, V: VectorIndex>(mut projector: Projector<E, V>, shared: Loc
                     delay
                 );
                 publish(&shared, projector.failure_status(&error, failures));
-                wait_for_work(&shared, Some(delay));
+                refuse_searches(&shared, &format!("the index could not advance: {error}"));
+                wait_for_work(&shared, Wait::Backoff(delay));
                 delay = backoff.after(delay);
             }
         }
@@ -538,7 +684,15 @@ impl WorkerSink {
         let mut projector = Projector::new(embedder, index, instance, mind);
         projector.want(startup);
         let shared: Lock = Arc::new((
-            Mutex::new(Shared { inbox: Vec::new(), mind: None, status: projector.progress_status(), closed: false }),
+            Mutex::new(Shared {
+                inbox: Vec::new(),
+                mind: None,
+                status: projector.progress_status(),
+                closed: false,
+                searches: VecDeque::new(),
+                found: Vec::new(),
+                next_ticket: 0,
+            }),
             Condvar::new(),
         ));
         let worker = Arc::clone(&shared);
@@ -596,6 +750,27 @@ impl<S: MindStore> IndexSink<S> for WorkerSink {
         let guard = lock(&self.shared);
         including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX))
     }
+
+    /// Queues the search for the worker and returns its ticket. A worker that
+    /// is failing or refused is not asked: its own report is the refusal.
+    fn search(&mut self, text: &str, top_k: u32) -> Result<SearchTicket, String> {
+        let mut guard = lock(&self.shared);
+        match &guard.status {
+            IndexStatus::Failing { error, .. } => return Err(format!("the semantic index is failing: {error}")),
+            IndexStatus::Refused { reason, .. } => return Err(format!("the semantic index is refused: {reason}")),
+            _ => {}
+        }
+        let ticket = SearchTicket(guard.next_ticket);
+        guard.next_ticket += 1;
+        guard.searches.push_back((ticket, text.to_owned(), top_k));
+        drop(guard);
+        self.shared.1.notify_all();
+        Ok(ticket)
+    }
+
+    fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
+        std::mem::take(&mut lock(&self.shared).found)
+    }
 }
 
 /// Plain HTTP to one base address, in text: the adapters own what the text

crates/huginn-daemon/src/index/qdrant.rs

@@ -7,7 +7,7 @@ use std::time::Duration;
 use anyhow::{Context, Result, ensure};
 use serde_json::{Value, json};
 
-use super::{CollectionMeta, Described, Http, Method, Point, VectorIndex};
+use super::{CollectionMeta, Described, Hit, Http, Method, Point, VectorIndex};
 
 /// One Qdrant. Every path names a collection this organ was told to own; it
 /// never lists or touches another.
@@ -132,4 +132,20 @@ impl VectorIndex for QdrantIndex {
         )?;
         Ok(())
     }
+
+    fn search(&mut self, collection: &str, vector: &[f32], limit: u32) -> Result<Vec<Hit>> {
+        let body = json!({ "vector": vector, "limit": limit, "with_payload": ["doc_id", "kind"], "with_vector": false });
+        let (_, answer) = self.ask(Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?;
+        let found = answer["result"].as_array().context("Qdrant's search has no result")?;
+        found
+            .iter()
+            .map(|point| {
+                Ok(Hit {
+                    doc_id: point["payload"]["doc_id"].as_str().context("a hit carries no doc_id")?.to_owned(),
+                    kind: point["payload"]["kind"].as_str().context("a hit carries no kind")?.to_owned(),
+                    score: point["score"].as_f64().context("a hit carries no score")? as f32,
+                })
+            })
+            .collect()
+    }
 }

crates/huginn-daemon/src/index/tests.rs

@@ -246,7 +246,7 @@ fn a_vector_of_the_wrong_length_is_refused_and_nothing_is_dropped() {
 }
 
 fn whoami_index(daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, WorkerSink>) -> IndexStatus {
-    match daemon.handle(HuginnMindRequest::Whoami, now()) {
+    match daemon.handle(HuginnMindRequest::Whoami, now()).answered() {
         HuginnMindResponse::Whoami(status) => status.index,
         other => panic!("expected a status, got {other:?}"),
     }
@@ -260,7 +260,7 @@ fn committed(response: HuginnMindResponse) {
 }
 
 fn quick() -> Backoff {
-    Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(2) }
+    Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(2), recheck: Duration::from_secs(3600) }
 }
 
 /// The index never delays or refuses admission: the embedder is shut, the
@@ -278,10 +278,10 @@ fn an_admission_replies_while_the_embedder_is_blocked() {
     let mut daemon = Daemon::new(mind, sink);
     daemon.index().wait_status(|status| *status == IndexStatus::Current);
 
-    committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, campaign_seed())), now()));
+    committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, campaign_seed())), now()).answered());
     gate.wait_arrived(1);
     let (question, ruling) = question_and_ruling();
-    committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now()));
+    committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now()).answered());
     assert_eq!(
         whoami_index(&mut daemon),
         IndexStatus::Behind { pending: 4 },
@@ -309,7 +309,7 @@ fn a_failing_embedder_is_visible_in_whoami_and_retried_never_dropped() {
     embedder.state.lock().unwrap().down = true;
     let (question, ruling) = question_and_ruling();
     for documents in [campaign_seed(), vec![question, ruling]] {
-        committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), now()));
+        committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), now()).answered());
     }
     let seen = daemon.index().wait_status(|status| matches!(status, IndexStatus::Failing { attempts, .. } if *attempts >= 2));
     let IndexStatus::Failing { pending, error, .. } = seen else { panic!() };
@@ -627,6 +627,7 @@ fn dropping_the_sink_stops_the_worker() {
 fn the_backoff_doubles_from_thirty_seconds_to_a_ten_minute_cap() {
     let backoff = Backoff::default();
     assert_eq!((backoff.initial, backoff.max), (Duration::from_secs(30), Duration::from_secs(600)));
+    assert_eq!(backoff.recheck, Duration::from_secs(60), "an idle worker verifies the model and the collection each minute");
     let mut delay = backoff.initial;
     let mut seen = vec![delay.as_secs()];
     for _ in 0..7 {
@@ -637,7 +638,7 @@ fn the_backoff_doubles_from_thirty_seconds_to_a_ten_minute_cap() {
         seen.push(delay.as_secs());
     }
     assert_eq!(seen, [30, 60, 120, 240, 480, 600, 600, 600]);
-    let small = Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(3) };
+    let small = Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(3), recheck: Duration::from_secs(3600) };
     assert_eq!(small.after(Duration::from_millis(1)), Duration::from_millis(2));
     assert_eq!(small.after(Duration::from_millis(2)), Duration::from_millis(3));
 }
@@ -666,8 +667,9 @@ fn a_collection_lost_after_startup_is_recreated_and_refilled() {
     assert_eq!(projector.progress_status(), IndexStatus::Current);
 }
 
-/// F2. A model re-pulled with other dimensions fails one write, then the next
-/// attempt reads the new identity and rebuilds the collection to match.
+/// F2. A model re-pulled with other dimensions is seen before anything is
+/// written: no vector of the new length is offered to the old collection, and
+/// the next step rebuilds the collection to match.
 #[test]
 fn a_model_that_changes_dimensions_after_startup_rebuilds_the_collection() {
     let (_root, mind) = mind_with_documents();
@@ -677,7 +679,9 @@ fn a_model_that_changes_dimensions_after_startup_rebuilds_the_collection() {
 
     embedder.state.lock().unwrap().identity.dimensions = 8;
     projector.want(entries[3..].iter().cloned());
-    assert!(projector.advance().is_err(), "a vector of 8 does not fit a collection of 4");
+    let writes = index.state.lock().unwrap().upserts.len();
+    assert_eq!(projector.advance().unwrap(), Advance::Progressed, "the change is seen, not written through");
+    assert_eq!(index.state.lock().unwrap().upserts.len(), writes, "no vector of 8 was offered to a collection of 4");
     drain(&mut projector);
 
     let state = index.state.lock().unwrap();
@@ -740,7 +744,7 @@ fn a_mind_without_an_identity_makes_no_collection_until_its_first_admission() {
     let mut daemon = Daemon::new(mind, sink);
     let seed = campaign_seed();
     let without_text = vec![seed[0].clone(), seed[1].clone()];
-    committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, without_text)), now()));
+    committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, without_text)), now()).answered());
     for _ in 0..500 {
         if index.state.lock().unwrap().collections.contains_key(&collection()) {
             break;
@@ -751,3 +755,234 @@ fn a_mind_without_an_identity_makes_no_collection_until_its_first_admission() {
     let Described::Labelled(label) = &state.collections[&collection()].label else { panic!("no collection was made") };
     assert_eq!(Some(&label.mind), daemon.mind().genesis_receipt_id().unwrap().as_ref());
 }
+
+/// Waits for a condition the worker will make true, without a clock in the
+/// assertion: it polls for at most ten seconds and then says what never
+/// happened.
+fn eventually(what: &str, condition: impl Fn() -> bool) {
+    for _ in 0..2000 {
+        if condition() {
+            return;
+        }
+        std::thread::sleep(Duration::from_millis(5));
+    }
+    panic!("{what} never happened");
+}
+
+/// A model re-pulled under the same name and dimensions but a new digest must
+/// not mix its vectors with the old model's. The check is before the write,
+/// not after a failure: nothing is offered to the old collection, the
+/// collection is rebuilt under the new digest, and every document is embedded
+/// again.
+#[test]
+fn a_model_re_pulled_with_the_same_dimensions_never_writes_beside_the_old_vectors() {
+    let (_root, mind) = mind_with_documents();
+    let entries = mind.index_entries(None).unwrap();
+    let (embedder, index) = pair();
+    let mut projector = projected(&embedder, &index, &entries[..3]);
+
+    embedder.state.lock().unwrap().identity.digest = "d2".into();
+    projector.want(entries[3..].iter().cloned());
+    let writes = index.state.lock().unwrap().upserts.len();
+    assert_eq!(projector.advance().unwrap(), Advance::Progressed);
+    assert_eq!(index.state.lock().unwrap().upserts.len(), writes, "the new vector was not written into the old collection");
+    assert_eq!(projector.progress_status(), IndexStatus::Reconciling { pending: 1 }, "the reconciliation was forgotten, not the entry");
+
+    drain(&mut projector);
+    let state = index.state.lock().unwrap();
+    assert_eq!(state.recreated.len(), 2);
+    assert_eq!(state.recreated.last().unwrap().1.model_digest, "d2");
+    let stored = &state.collections[&collection()];
+    assert_eq!(stored.points.len(), 4, "every document is in the rebuilt collection");
+    assert_eq!(embedder.texts().len(), 3 + 4, "and was embedded again under the new digest");
+}
+
+/// Nothing is written while idle, so no write can fail to announce that the
+/// collection is gone or the model has changed. The idle worker looks again on
+/// its own clock: a lost collection is refilled and a re-pulled model rebuilds
+/// it, with no admission to prompt either.
+#[test]
+fn an_idle_worker_finds_a_lost_collection_and_a_changed_model_without_a_write() {
+    let (_root, mind) = mind_with_documents();
+    let entries = mind.index_entries(None).unwrap();
+    let (embedder, index) = pair();
+    let recheck = Backoff { recheck: Duration::from_millis(20), ..quick() };
+    let sink = WorkerSink::spawn(embedder.clone(), index.clone(), &slug(INSTANCE), mind.genesis_receipt_id().unwrap(), entries.clone(), recheck);
+    sink.wait_status(|status| *status == IndexStatus::Current);
+    assert_eq!(index.holding(&collection()), point_ids(&entries));
+
+    index.state.lock().unwrap().collections.remove(&collection());
+    eventually("the lost collection being made again and refilled", || {
+        index.state.lock().unwrap().recreated.len() == 2 && index.holding(&collection()) == point_ids(&entries)
+    });
+
+    embedder.state.lock().unwrap().identity.digest = "d2".into();
+    eventually("the collection being rebuilt under the new digest", || {
+        let state = index.state.lock().unwrap();
+        state.recreated.len() == 3
+            && state.recreated.last().is_some_and(|(_, meta)| meta.model_digest == "d2")
+            && state.collections[&collection()].points.len() == 4
+    });
+    sink.wait_status(|status| *status == IndexStatus::Current);
+}
+
+/// 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.
+#[test]
+fn a_search_embeds_the_query_with_its_instruction_and_oversamples() {
+    let (_root, mind) = mind_with_documents();
+    let entries = mind.index_entries(None).unwrap();
+    let (embedder, index) = pair();
+    let mut projector = projected(&embedder, &index, &entries);
+    index.state.lock().unwrap().hits = vec![
+        Hit { doc_id: "eureka-state:question:Q1".into(), kind: "question".into(), score: 0.9 },
+        Hit { doc_id: "eureka-state:mystery:X".into(), kind: "mystery".into(), score: 0.8 },
+    ];
+
+    let hits = projector.search("who owns the state", 3).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).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),
+    ]);
+}
+
+/// A search is never run against a collection that is not known to fit the
+/// model: before reconciliation, and after the model has changed, it refuses
+/// by saying so rather than comparing vectors of two models.
+#[test]
+fn a_search_refuses_when_the_index_is_not_reconciled_with_the_model() {
+    let (_root, mind) = mind_with_documents();
+    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).unwrap_err()).contains("not been reconciled"));
+    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).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
+/// 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() {
+    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).unwrap();
+    assert_eq!(
+        hits,
+        [
+            Hit { doc_id: "eureka-state:question:Q1".into(), kind: "question".into(), score: 0.75 },
+            Hit { doc_id: "eureka-state:ruling:R1".into(), kind: "ruling".into(), score: 0.5 },
+        ]
+    );
+    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);
+    drop(seen);
+
+    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).is_err());
+    let failing = stub(&[("POST /collections/", 404, r#"{"status":{"error":"not found"}}"#)]);
+    assert!(QdrantIndex::new(&failing.url).search(&name, &[1.0], 1).is_err());
+}
+
+fn shared_with(
+    inbox: Vec<IndexEntry>,
+    mind: Option<String>,
+    searches: Vec<(SearchTicket, String, u32)>,
+    closed: bool,
+) -> Lock {
+    Arc::new((
+        Mutex::new(Shared {
+            inbox,
+            mind,
+            status: IndexStatus::Current,
+            closed,
+            searches: searches.into(),
+            found: Vec::new(),
+            next_ticket: 0,
+        }),
+        Condvar::new(),
+    ))
+}
+
+/// The idle wait is for work: an entry, the mind's identity, a search or the
+/// sink closing ends it at once, and only a quiet worker waits out the recheck
+/// interval, after which it says the interval passed.
+#[test]
+fn an_idle_wait_ends_for_work_and_otherwise_at_the_recheck() {
+    let (_root, mind) = mind_with_documents();
+    let entry = mind.index_entries(None).unwrap().remove(0);
+    let waits = |shared: &Lock| {
+        let started = Instant::now();
+        let recheck = wait_for_work(shared, Wait::Work { recheck: Duration::from_secs(30) });
+        (recheck, started.elapsed() < Duration::from_secs(10))
+    };
+    assert_eq!(waits(&shared_with(vec![entry], None, Vec::new(), false)), (false, true), "an entry");
+    assert_eq!(waits(&shared_with(Vec::new(), Some(MIND.into()), Vec::new(), false)), (false, true), "the mind's identity");
+    assert_eq!(waits(&shared_with(Vec::new(), None, vec![(SearchTicket(0), "q".into(), 3)], false)), (false, true), "a search");
+    assert_eq!(waits(&shared_with(Vec::new(), None, Vec::new(), true)), (false, true), "the sink closing");
+
+    let quiet = shared_with(Vec::new(), None, Vec::new(), false);
+    let started = Instant::now();
+    assert!(wait_for_work(&quiet, Wait::Work { recheck: Duration::from_millis(60) }), "a quiet worker is told the interval passed");
+    assert!(started.elapsed() >= Duration::from_millis(60), "and only after it did");
+}
+
+/// A backoff is the whole delay, whatever arrives, and never reports a
+/// recheck: a failing index is retried on its schedule.
+#[test]
+fn a_backoff_wait_is_the_whole_delay_whatever_arrives() {
+    let (_root, mind) = mind_with_documents();
+    let entry = mind.index_entries(None).unwrap().remove(0);
+    let busy = shared_with(vec![entry], Some(MIND.into()), vec![(SearchTicket(0), "q".into(), 3)], false);
+    let started = Instant::now();
+    assert!(!wait_for_work(&busy, Wait::Backoff(Duration::from_millis(80))));
+    assert!(started.elapsed() >= Duration::from_millis(80), "the delay was waited out although work was queued");
+    let closed = shared_with(Vec::new(), None, Vec::new(), true);
+    let started = Instant::now();
+    assert!(!wait_for_work(&closed, Wait::Backoff(Duration::from_secs(30))));
+    assert!(started.elapsed() < Duration::from_secs(10), "a closed sink ends even a backoff");
+}
+
+/// Searches accepted while the index was healthy are answered, with the
+/// reason, when the worker fails: nothing is left to wait for its deadline.
+#[test]
+fn queued_searches_are_answered_with_why_when_the_worker_gives_up() {
+    let queued = vec![(SearchTicket(3), "a".into(), 1), (SearchTicket(4), "b".into(), 2)];
+    let shared = shared_with(Vec::new(), None, queued, false);
+    refuse_searches(&shared, "the index could not advance: down");
+    let guard = lock(&shared);
+    assert!(guard.searches.is_empty());
+    assert_eq!(
+        guard.found,
+        [
+            (SearchTicket(3), Err("the index could not advance: down".to_string())),
+            (SearchTicket(4), Err("the index could not advance: down".to_string())),
+        ]
+    );
+}

crates/huginn-daemon/src/lib.rs

@@ -21,6 +21,9 @@ pub mod envelope;
 pub mod index;
 pub mod serve;
 
-pub use daemon::{Daemon, IndexSink, runtime_id};
+#[cfg(test)]
+mod semantic_tests;
+
+pub use daemon::{Daemon, Handled, Hits, IndexSink, Search, SearchTicket, runtime_id};
 pub use envelope::{FAILURE_SCHEMA, OperationFailure};
 pub use serve::{Options, ServeOptions, parse_options, run, startup};

crates/huginn-daemon/src/semantic_tests.rs

@@ -0,0 +1,434 @@
+//! The semantic read through the real serve loop: a UDP hub, real sessions, the
+//! worker thread, and fake embedder and vector-store ports whose behaviour the
+//! test decides. Nothing here calls `Daemon::handle` directly: the rules under
+//! test are about what the loop does while an answer is outstanding.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread::JoinHandle;
+use std::time::Duration;
+
+use anyhow::Result;
+use cultnet_rs::{CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, FieldPredicate, Selection};
+use huginn_mind::epiphany_pipeline::{
+    Date, PipelineDocument, PipelineKind, PipelineQuestion, PipelineResolution, PipelineRef, PipelineRuling,
+    QuestionOption, ResolutionOutcome, RulingAuthority, Short, Title,
+};
+use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, IndexStatus};
+use huginn_mind::{Mind, MindRefusal, OwnedRedbMessagePackBackingStore, PipelineAdmissionOutcome, PipelinePageItems, SemanticQuery};
+use tempfile::TempDir;
+
+use crate::daemon::Daemon;
+use crate::daemon::tests::{CAMPAIGN, INSTANCE, batch, campaign_seed, now, slug};
+use crate::envelope::{decode_response, encode_request};
+use crate::index::fakes::{FakeEmbedder, FakeIndex, Gate};
+use crate::index::{Backoff, Described, Hit, WorkerSink, collection_name};
+use crate::serve::{ServeOptions, bind, run, schema_registry};
+
+type Client = cultnet_rs::CultNetRudpSocketTransportConnection;
+
+fn question(label: &str) -> PipelineDocument {
+    PipelineDocument::Question(PipelineQuestion {
+        campaign: slug(CAMPAIGN),
+        label: label.into(),
+        title: Title(format!("Question {label}")),
+        question: format!("Question {label}?").as_str().into(),
+        options: vec![
+            QuestionOption { label: "A".into(), text: "the organ".into() },
+            QuestionOption { label: "B".into(), text: "the client".into() },
+        ],
+        recommended: "A".into(),
+        depends: vec![],
+        raised_in: None,
+        asked_on: Date("2026-09-16".into()),
+    })
+}
+
+/// A ruling, answering a question when it names one.
+fn ruling(label: &str, answers: Option<&str>) -> PipelineDocument {
+    PipelineDocument::Ruling(PipelineRuling {
+        campaign: slug(CAMPAIGN),
+        label: label.into(),
+        title: Title(format!("Ruling {label}")),
+        answers: answers.map(|question| Short(id("question", question))),
+        choice: answers.map(|_| "A".into()),
+        ruling: format!("Ruling {label}.").as_str().into(),
+        operator_quote: None,
+        ruled_on: Date("2026-09-16".into()),
+        precedents: vec![],
+        authority: RulingAuthority::Operator,
+    })
+}
+
+fn resolution(subject: PipelineKind, key: &str, outcome: ResolutionOutcome) -> PipelineDocument {
+    PipelineDocument::Resolution(PipelineResolution {
+        subject: PipelineRef { kind: subject, id: Short(key.into()) },
+        sequence: 1,
+        outcome,
+        rationale: "Resolved.".into(),
+        resolved_on: Date("2026-09-16".into()),
+    })
+}
+
+fn id(kind: &str, local: &str) -> String {
+    format!("{CAMPAIGN}:{kind}:{local}")
+}
+
+fn hit(kind: &str, key: String, score: f32) -> Hit {
+    Hit { doc_id: key, kind: kind.into(), score }
+}
+
+/// Every way a document stops standing, and two that do: Q1 is answered (so
+/// resolved), Q2 was answered and the answer withdrawn (its resolution is the
+/// resolved one and the question stands again), R3 is superseded by R4.
+fn shelf() -> Vec<Vec<PipelineDocument>> {
+    vec![
+        campaign_seed(),
+        vec![question("Q1"), question("Q2")],
+        vec![ruling("R1", Some("Q1"))],
+        vec![ruling("R2", Some("Q2"))],
+        vec![resolution(PipelineKind::Resolution, &id("resolution", "question.Q2.n1"), ResolutionOutcome::Withdrawn { reason: "moot".into() })],
+        vec![ruling("R3", None), ruling("R4", None)],
+        vec![resolution(
+            PipelineKind::Ruling,
+            &id("ruling", "R3"),
+            ResolutionOutcome::Superseded { by: vec![PipelineRef { kind: PipelineKind::Ruling, id: Short(id("ruling", "R4")) }] },
+        )],
+    ]
+}
+
+fn standing() -> Selection {
+    Selection {
+        fields: Some(vec![FieldPredicate {
+            index: "in_force".into(),
+            op: "any_of".into(),
+            values: Some(vec!["true".into()]),
+            number: None,
+        }]),
+        ..Selection::default()
+    }
+}
+
+fn semantic(selection: Selection, top_k: u32) -> HuginnMindRequest {
+    HuginnMindRequest::Query {
+        instance: slug(INSTANCE),
+        selection,
+        semantic: Some(SemanticQuery { text: "who owns the state".into(), top_k }),
+    }
+}
+
+/// How many documents a plain query finds.
+fn documents(client: &mut Client, message_id: &str) -> usize {
+    ids(&ask(client, message_id, &plain())).len()
+}
+
+fn plain() -> HuginnMindRequest {
+    HuginnMindRequest::Query { instance: slug(INSTANCE), selection: Selection::default(), semantic: None }
+}
+
+/// The ids on a page, in its order.
+fn ids(response: &HuginnMindResponse) -> Vec<String> {
+    let HuginnMindResponse::Query(page) = response else { panic!("expected a page, got {response:?}") };
+    match &page.items {
+        PipelinePageItems::Headers(headers) => headers.iter().map(|header| header.id.id.0.clone()).collect(),
+        PipelinePageItems::Documents(views) => views.iter().map(|view| view.id.id.0.clone()).collect(),
+    }
+}
+
+fn quick() -> Backoff {
+    Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(2), recheck: Duration::from_secs(3600) }
+}
+
+/// A daemon serving a mind over a real socket, with the ports the test gave it.
+struct Harness {
+    endpoint: String,
+    stopping: Arc<AtomicBool>,
+    serving: Option<JoinHandle<Result<()>>>,
+    embedder: FakeEmbedder,
+    index: FakeIndex,
+    _root: TempDir,
+}
+
+impl Harness {
+    fn start(seed: Vec<Vec<PipelineDocument>>, embedder: FakeEmbedder, index: FakeIndex, search_timeout: Duration) -> Self {
+        let root = tempfile::tempdir().unwrap();
+        let mut mind = Mind::open(root.path(), &slug(INSTANCE)).unwrap();
+        for documents in seed {
+            let outcome = mind.admit(batch(INSTANCE, documents), now());
+            assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
+        }
+        let sink = WorkerSink::spawn(
+            embedder.clone(),
+            index.clone(),
+            &slug(INSTANCE),
+            mind.genesis_receipt_id().unwrap(),
+            mind.index_entries(None).unwrap(),
+            quick(),
+        );
+        let mut daemon: Daemon<OwnedRedbMessagePackBackingStore, WorkerSink> = Daemon::new(mind, sink);
+        let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap();
+        let endpoint = format!("rudp://{}", hub.local_addr().unwrap());
+        let registry = schema_registry().unwrap();
+        let stopping = Arc::new(AtomicBool::new(false));
+        let loop_stopping = Arc::clone(&stopping);
+        let options = ServeOptions { search_timeout, ..ServeOptions::default() };
+        let serving = std::thread::spawn(move || run(&mut daemon, &mut hub, &registry, &loop_stopping, &options));
+        Self { endpoint, stopping, serving: Some(serving), embedder, index, _root: root }
+    }
+
+    fn client(&self) -> Client {
+        let mut client = CultMesh::create_rudp_client_for_endpoint(
+            "semantic-test".to_string(),
+            CULTNET_OPERATION_CONNECTION_ID,
+            &self.endpoint,
+            CultMeshRudpSocketOptions::default(),
+        )
+        .unwrap();
+        client.connect(Vec::new()).unwrap();
+        for _ in 0..1000 {
+            let _ = client.receive_once();
+            if client.connected() {
+                return client;
+            }
+            std::thread::sleep(Duration::from_millis(2));
+        }
+        panic!("the client never connected");
+    }
+}
+
+impl Drop for Harness {
+    fn drop(&mut self) {
+        self.stopping.store(true, Ordering::Relaxed);
+        // A loop that is stuck waiting on the index is a failure of the test,
+        // not a reason to hang the run: give it a moment and leave it behind.
+        if let Some(serving) = self.serving.take() {
+            for _ in 0..1500 {
+                if serving.is_finished() {
+                    let _ = serving.join();
+                    return;
+                }
+                std::thread::sleep(Duration::from_millis(2));
+            }
+        }
+    }
+}
+
+fn send(client: &mut Client, message_id: &str, request: &HuginnMindRequest) {
+    client.send_schema_message(&encode_request(message_id, request, None).unwrap()).unwrap();
+}
+
+/// The reply to `message_id` on this client's session, or `None` if none came
+/// in `patience` polls of two milliseconds.
+fn receive(client: &mut Client, message_id: &str, patience: u32) -> Option<HuginnMindResponse> {
+    for _ in 0..patience {
+        client.poll_resends().unwrap();
+        if let Some(reply) = client.receive_schema_message_once().unwrap() {
+            let (correlated, response) = decode_response(&reply).unwrap();
+            assert_eq!(correlated, message_id);
+            return Some(response.expect("an answer, not an envelope failure"));
+        }
+        std::thread::sleep(Duration::from_millis(2));
+    }
+    None
+}
+
+fn ask(client: &mut Client, message_id: &str, request: &HuginnMindRequest) -> HuginnMindResponse {
+    send(client, message_id, request);
+    receive(client, message_id, 3000).unwrap_or_else(|| panic!("no reply to {message_id}"))
+}
+
+fn index_status(client: &mut Client, message_id: &str) -> IndexStatus {
+    match ask(client, message_id, &HuginnMindRequest::Whoami) {
+        HuginnMindResponse::Whoami(status) => status.index,
+        other => panic!("expected a status, got {other:?}"),
+    }
+}
+
+fn until_index(client: &mut Client, what: &str, predicate: impl Fn(&IndexStatus) -> bool) {
+    for at in 0..2000 {
+        if predicate(&index_status(client, &format!("status-{at}"))) {
+            return;
+        }
+        std::thread::sleep(Duration::from_millis(5));
+    }
+    panic!("the index never reported {what}");
+}
+
+fn healthy() -> Harness {
+    let (embedder, index) = (FakeEmbedder::new("d1"), FakeIndex::default());
+    Harness::start(shelf(), embedder, index, Duration::from_secs(30))
+}
+
+/// A semantic answer is the mind's: the index names a resolved question, a
+/// withdrawn resolution, a superseded ruling and a document the mind does not
+/// 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.
+#[test]
+fn a_semantic_query_returns_only_documents_the_mind_holds_and_the_selection_admits() {
+    let harness = healthy();
+    let mut client = harness.client();
+    until_index(&mut client, "Current", |status| *status == IndexStatus::Current);
+    harness.index.state.lock().unwrap().hits = vec![
+        hit("question", id("question", "Q1"), 0.95),
+        hit("resolution", id("resolution", "question.Q2.n1"), 0.9),
+        hit("ruling", id("ruling", "R3"), 0.85),
+        hit("question", id("question", "Q99"), 0.8),
+        hit("ruling", id("ruling", "R4"), 0.5),
+        hit("question", id("question", "Q2"), 0.4),
+    ];
+
+    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 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");
+
+    let everything = ids(&ask(&mut client, "m-3", &semantic(Selection::default(), 5)));
+    assert!(everything.contains(&id("question", "Q1")) && !everything.contains(&id("question", "Q99")),
+        "a hit the selection does not exclude is shown, and one the mind does not hold never is: {everything:?}");
+
+    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:?}");
+}
+
+/// A cursor cannot resume a ranked answer: it is refused by name, and nothing
+/// is embedded or searched for it.
+#[test]
+fn a_cursor_with_a_semantic_query_is_refused_and_asks_the_index_nothing() {
+    let harness = healthy();
+    let mut client = harness.client();
+    until_index(&mut client, "Current", |status| *status == IndexStatus::Current);
+    let embedded = harness.embedder.texts().len();
+
+    let with_cursor = Selection { cursor: Some("anything".into()), ..standing() };
+    let refusal = ask(&mut client, "m-1", &semantic(with_cursor, 5));
+    assert!(
+        matches!(&refusal, HuginnMindResponse::Refused(MindRefusal::SelectionInvalid { field, .. }) if field == "cursor"),
+        "{refusal:?}"
+    );
+    assert_eq!(harness.embedder.texts().len(), embedded);
+    assert!(harness.index.state.lock().unwrap().searches.is_empty());
+}
+
+/// The loop never waits on the network. The embedder is shut, the worker is
+/// inside it with one client's semantic query, and a second client's `whoami`
+/// and plain query are answered at once; the first is answered when the gate
+/// opens. The gate counts arrivals, so this observes where the work is and not
+/// how long it took.
+#[test]
+fn another_session_is_answered_while_a_semantic_search_is_outstanding() {
+    let gate = Gate::shut();
+    gate.open();
+    let embedder = FakeEmbedder::new("d1").behind(&gate);
+    let harness = Harness::start(shelf(), embedder, FakeIndex::default(), Duration::from_secs(30));
+    let mut asker = harness.client();
+    let mut second = harness.client();
+    let mut other = harness.client();
+    until_index(&mut other, "Current", |status| *status == IndexStatus::Current);
+    let held = documents(&mut other, "m-count");
+    harness.index.state.lock().unwrap().hits = vec![hit("ruling", id("ruling", "R4"), 0.5)];
+
+    gate.shut_again();
+    send(&mut asker, "m-ask", &semantic(standing(), 5));
+    gate.wait_arrived(1);
+    assert!(receive(&mut asker, "m-ask", 20).is_none(), "the question is not answered while its embedding is outstanding");
+    // A second question waits behind the first, and each is answered to its own asker.
+    send(&mut second, "m-ask-2", &semantic(Selection { schemas: Some(vec![PipelineKind::Question.type_id().into()]), ..Selection::default() }, 5));
+
+    assert!(matches!(ask(&mut other, "m-who", &HuginnMindRequest::Whoami), HuginnMindResponse::Whoami(_)));
+    assert_eq!(documents(&mut other, "m-plain"), held, "a plain query is answered too");
+
+    gate.open();
+    let answered = receive(&mut asker, "m-ask", 3000).expect("the search is answered once the embedder returns");
+    assert_eq!(ids(&answered), vec![id("ruling", "R4")]);
+    let second_answer = receive(&mut second, "m-ask-2", 3000).expect("the second search is answered too, on its own session");
+    assert_eq!(ids(&second_answer), Vec::<String>::new(), "the index offered a ruling, which its selection of questions does not admit");
+}
+
+/// A search the index never finishes is answered at the deadline the client
+/// was promised, typed and naming the wait, never as an empty page.
+#[test]
+fn a_search_the_index_never_finishes_is_answered_unavailable_at_the_deadline() {
+    let gate = Gate::shut();
+    gate.open();
+    let embedder = FakeEmbedder::new("d1").behind(&gate);
+    let harness = Harness::start(shelf(), embedder, FakeIndex::default(), Duration::from_millis(150));
+    let mut client = harness.client();
+    until_index(&mut client, "Current", |status| *status == IndexStatus::Current);
+
+    gate.shut_again();
+    let answered = ask(&mut client, "m-late", &semantic(standing(), 5));
+    let HuginnMindResponse::Refused(MindRefusal::Unavailable { detail }) = &answered else { panic!("{answered:?}") };
+    assert!(detail.contains("did not finish"), "{detail}");
+    gate.open();
+    assert!(matches!(ask(&mut client, "m-after", &HuginnMindRequest::Whoami), HuginnMindResponse::Whoami(_)), "the late answer is dropped, not misdelivered");
+}
+
+/// An index that cannot answer is a typed `Unavailable` and never an empty
+/// page, in each way it cannot: the embedder is down when the query arrives;
+/// the worker is failing; the collection belongs to someone else. Plain
+/// queries are unaffected throughout.
+#[test]
+fn an_index_that_cannot_answer_is_unavailable_and_plain_queries_still_work() {
+    let unavailable = |response: HuginnMindResponse| match response {
+        HuginnMindResponse::Refused(MindRefusal::Unavailable { detail }) => detail,
+        other => panic!("expected Unavailable, got {other:?}"),
+    };
+
+    // The embedder goes down after the index was Current: the query itself fails.
+    let harness = healthy();
+    let mut client = harness.client();
+    until_index(&mut client, "Current", |status| *status == IndexStatus::Current);
+    let held = documents(&mut client, "m-count");
+    harness.embedder.state.lock().unwrap().down = true;
+    let detail = unavailable(ask(&mut client, "m-down", &semantic(standing(), 5)));
+    assert!(detail.contains("the embedder is down"), "{detail}");
+    assert_eq!(documents(&mut client, "m-plain"), held);
+
+    // A write while it is down makes the worker Failing, and the next query is
+    // refused without being asked of it.
+    let admit = HuginnMindRequest::Admit(batch(INSTANCE, vec![question("Q7")]));
+    assert!(matches!(ask(&mut client, "m-admit", &admit), HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })));
+    until_index(&mut client, "Failing", |status| matches!(status, IndexStatus::Failing { .. }));
+    let searches = harness.index.state.lock().unwrap().searches.len();
+    let detail = unavailable(ask(&mut client, "m-failing", &semantic(standing(), 5)));
+    assert!(detail.contains("failing") && detail.contains("the embedder is down"), "{detail}");
+    assert_eq!(harness.index.state.lock().unwrap().searches.len(), searches);
+    assert_eq!(documents(&mut client, "m-plain-2"), held + 1, "the admitted question is in the mind whatever the index says");
+
+    // A collection that is another writer's is Refused, and so are queries.
+    let (embedder, index) = (FakeEmbedder::new("d1"), FakeIndex::default());
+    index.plant(&collection_name(&slug(INSTANCE)), Described::Unlabelled);
+    let refused = Harness::start(shelf(), embedder, index, Duration::from_secs(30));
+    let mut client = refused.client();
+    until_index(&mut client, "Refused", |status| matches!(status, IndexStatus::Refused { .. }));
+    let detail = unavailable(ask(&mut client, "m-refused", &semantic(standing(), 5)));
+    assert!(detail.contains("refused") && detail.contains("no Huginn metadata"), "{detail}");
+    assert_eq!(documents(&mut client, "m-plain-3"), held);
+}
+
+/// The ids a daemon reports as unheld are exactly the hits its mind has no
+/// document for: an absent id, and an id under a kind it is not.
+#[test]
+fn a_hit_the_mind_holds_no_document_for_is_named_unheld() {
+    let root = tempfile::tempdir().unwrap();
+    let mut daemon = crate::daemon::tests::open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
+    let identity = crate::daemon::tests::identity(INSTANCE);
+    let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) =
+        daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity])), now()).answered()
+    else {
+        panic!("the identity was not admitted")
+    };
+    let held = writes[0].clone();
+    let absent = PipelineRef { kind: held.kind, id: Short(id("instance", "nobody")) };
+    let mis_kinded = PipelineRef { kind: PipelineKind::Question, id: held.id.clone() };
+    let hits = vec![(held.clone(), 0.9), (absent.clone(), 0.8), (mis_kinded.clone(), 0.7)];
+    assert_eq!(daemon.unheld(&hits), vec![absent.id.0.as_str(), mis_kinded.id.0.as_str()]);
+    assert!(daemon.unheld(&vec![(held, 0.9)]).is_empty());
+}

crates/huginn-daemon/src/serve.rs

@@ -56,7 +56,7 @@ use std::collections::BTreeMap;
 use std::net::{SocketAddr, UdpSocket};
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use anyhow::{Context, Result, ensure};
 use base64::Engine;
@@ -64,7 +64,7 @@ use base64::engine::general_purpose::STANDARD;
 use chrono::{DateTime, Utc};
 use cultnet_rs::{
     CULTNET_OPERATION_CONNECTION_ID, CultNetMessage, CultNetRudpServerEvent, CultNetRudpServerHub,
-    CultNetRudpServerHubOptions, CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry,
+    CultNetRudpServerHubOptions, CultNetRudpServerSessionContext, CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry,
     CultNetWireContract, answer_content_chunk_request, decode_cultnet_message_from_slice,
     encode_cultnet_message_to_vec, pack_content,
 };
@@ -78,7 +78,7 @@ use huginn_mind::{Mind, MindRefusal, MindStore, OwnedRedbMessagePackBackingStore
 use crate::bodies::{
     DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies, MAX_DEFERRED_BODY_BYTES,
 };
-use crate::daemon::{Daemon, IndexSink, runtime_id};
+use crate::daemon::{Daemon, Handled, IndexSink, Search, SearchTicket, runtime_id};
 use crate::envelope::{OperationFailure, decode_request, encode_failure, encode_response};
 use crate::index::{Backoff, Embedder, VectorIndex, WorkerSink};
 
@@ -88,6 +88,10 @@ pub struct ServeOptions {
     pub session_timeout: Duration,
     pub idle_sleep: Duration,
     pub deferred_ttl: Duration,
+    /// How long a semantic query may wait for the index before it is answered
+    /// `Unavailable`. The worker's own calls time out on their own; this is
+    /// what the client is promised.
+    pub search_timeout: Duration,
 }
 
 impl Default for ServeOptions {
@@ -96,6 +100,7 @@ impl Default for ServeOptions {
             session_timeout: Duration::from_secs(30),
             idle_sleep: Duration::from_millis(2),
             deferred_ttl: Duration::from_secs(60),
+            search_timeout: Duration::from_secs(30),
         }
     }
 }
@@ -250,33 +255,54 @@ fn overlong_echo(message: &CultNetMessage) -> Option<&'static str> {
     echoed.into_iter().find(|(_, value)| value.len() > MAX_ECHOED_FIELD_BYTES).map(|(name, _)| name)
 }
 
-/// One message in, one message out. Nothing here reaches a mind except through
-/// `Daemon::handle`, and an envelope that does not decode never gets that far.
-/// A chunk request is answered from `bodies` alone, on the session it came in
-/// on; every operation's reply leaves through `deliver`.
+/// What one message came to: a message to send back now, or a semantic query
+/// the index has yet to answer, which the loop holds against the session it
+/// came in on.
+pub enum Routed {
+    Reply(CultNetMessage),
+    Search { message_id: String, operation: &'static str, search: Search },
+}
+
+impl Routed {
+    /// The reply, for a message that cannot be a search.
+    #[cfg(test)]
+    pub(crate) fn reply(self) -> CultNetMessage {
+        match self {
+            Self::Reply(reply) => reply,
+            Self::Search { .. } => panic!("expected a reply, got a search"),
+        }
+    }
+}
+
+/// One message in, one message out, or a search to wait for. Nothing here
+/// reaches a mind except through `Daemon::handle`, and an envelope that does
+/// not decode never gets that far. A chunk request is answered from `bodies`
+/// alone, on the session it came in on; every operation's reply leaves through
+/// `respond`.
 pub fn answer<S: MindStore, I: IndexSink<S>>(
     daemon: &mut Daemon<S, I>,
     registry: &CultNetSchemaRegistry,
     bodies: &mut DeferredBodies,
     message: CultNetMessage,
     now: DateTime<Utc>,
-) -> CultNetMessage {
+) -> Routed {
     let runtime_id = daemon.runtime_id();
     bodies.expire(now);
     if let Some(field) = overlong_echo(&message) {
-        return CultNetMessage::Error {
+        return Routed::Reply(CultNetMessage::Error {
             error: format!("{MIND_SERVICE_ID} refuses a {field} over {MAX_ECHOED_FIELD_BYTES} bytes"),
             code: None,
             details: None,
-        };
+        });
     }
-    match &message {
+    Routed::Reply(match &message {
         CultNetMessage::OperationRequest { message_id, operation, .. } => match decode_request(&message) {
             Ok((message_id, request)) => {
                 let operation = request.operation();
-                let response = daemon.handle(request, now);
-                let reply = encode_or_fail(&message_id, operation, &response, &runtime_id);
-                deliver(reply, &message_id, operation, &runtime_id, bodies, now)
+                match daemon.handle(request, now) {
+                    Handled::Answered(response) => respond(&response, &message_id, operation, &runtime_id, bodies, now),
+                    Handled::Searching(search) => return Routed::Search { message_id, operation, search },
+                }
             }
             Err(failure) => encode_failure(message_id, operation, &failure, &runtime_id),
         },
@@ -295,7 +321,21 @@ pub fn answer<S: MindStore, I: IndexSink<S>>(
             code: None,
             details: None,
         },
-    }
+    })
+}
+
+/// One answer as the message that carries it: encoded, then delivered whole,
+/// deferred or refused by `deliver`.
+fn respond(
+    response: &HuginnMindResponse,
+    message_id: &str,
+    operation: &str,
+    runtime_id: &str,
+    bodies: &mut DeferredBodies,
+    now: DateTime<Utc>,
+) -> CultNetMessage {
+    let reply = encode_or_fail(message_id, operation, response, runtime_id);
+    deliver(reply, message_id, operation, runtime_id, bodies, now)
 }
 
 /// One response in its envelope, or the failure that says it did not encode.
@@ -397,10 +437,24 @@ fn deliver(
     deferred
 }
 
+/// A semantic query on its way through the index: who asked, what to call the
+/// answer, the search to finish, and when the client stops being promised one.
+struct Waiting {
+    session: CultNetRudpServerSessionContext,
+    message_id: String,
+    operation: &'static str,
+    search: Search,
+    until: Instant,
+}
+
 /// Until `stopping`: expire what timed out, resend what was not acknowledged,
-/// then answer every frame waiting on the socket. A hostile datagram and a
-/// departed session are logged and served past, never fatal. This is the
-/// crate's only clock read, once per frame.
+/// answer every frame waiting on the socket, then answer every semantic query
+/// the index has finished or that has waited out `search_timeout`. A semantic
+/// query is never answered inline: `answer` hands back a `Search`, the loop
+/// holds it beside its session and goes on serving, and the embedder and the
+/// vector store are the worker's to wait on. A hostile datagram and a departed
+/// session are logged and served past, never fatal. The clock is read once per
+/// frame, and once per pass for the searches' deadlines.
 pub fn run<S: MindStore, I: IndexSink<S>>(
     daemon: &mut Daemon<S, I>,
     hub: &mut CultNetRudpServerHub,
@@ -410,6 +464,7 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
 ) -> Result<()> {
     let timeout_ms = options.session_timeout.as_millis() as u64;
     let mut bodies = DeferredBodies::new(DEFERRED_BUDGET_BYTES, options.deferred_ttl);
+    let mut waiting: BTreeMap<SearchTicket, Waiting> = BTreeMap::new();
     while !stopping.load(Ordering::Relaxed) {
         hub.remove_timed_out_sessions(timeout_ms);
         if let Err(error) = hub.poll_resends() {
@@ -439,19 +494,54 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
                     continue;
                 }
             };
-            let reply = answer(daemon, registry, &mut bodies, message, Utc::now());
-            if let Err(error) = hub.send_schema_message(&session, &reply) {
-                eprintln!("huginn: {} did not receive its reply: {error:#}", session.remote_addr);
+            match answer(daemon, registry, &mut bodies, message, Utc::now()) {
+                Routed::Reply(reply) => send(hub, &session, &reply),
+                Routed::Search { message_id, operation, search } => {
+                    let until = Instant::now() + options.search_timeout;
+                    waiting.insert(search.ticket, Waiting { session, message_id, operation, search, until });
+                }
             }
             served += 1;
         }
-        if served == 0 {
+        let finished = collect_searches(daemon, &waiting, options);
+        let answered = !finished.is_empty();
+        for (ticket, found) in finished {
+            let Some(asked) = waiting.remove(&ticket) else { continue };
+            let response = daemon.finish(asked.search, found);
+            let reply = respond(&response, &asked.message_id, asked.operation, &daemon.runtime_id(), &mut bodies, Utc::now());
+            send(hub, &asked.session, &reply);
+        }
+        if served == 0 && !answered {
             std::thread::sleep(options.idle_sleep);
         }
     }
     Ok(())
 }
 
+/// What the index has finished, then what has waited too long: the latter as
+/// an `Unavailable` detail. A search that finishes after its deadline finds
+/// nothing waiting for it and is dropped.
+fn collect_searches<S: MindStore, I: IndexSink<S>>(
+    daemon: &mut Daemon<S, I>,
+    waiting: &BTreeMap<SearchTicket, Waiting>,
+    options: &ServeOptions,
+) -> Vec<(SearchTicket, Result<crate::daemon::Hits, String>)> {
+    let mut collected = daemon.searched();
+    let now = Instant::now();
+    for (ticket, asked) in waiting {
+        if asked.until <= now {
+            collected.push((*ticket, Err(format!("the semantic search did not finish within {:?}", options.search_timeout))));
+        }
+    }
+    collected
+}
+
+fn send(hub: &mut CultNetRudpServerHub, session: &CultNetRudpServerSessionContext, reply: &CultNetMessage) {
+    if let Err(error) = hub.send_schema_message(session, reply) {
+        eprintln!("huginn: {} did not receive its reply: {error:#}", session.remote_addr);
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -472,7 +562,7 @@ mod tests {
     use std::sync::atomic::AtomicBool;
 
     fn documents(daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, NoIndex>) -> u32 {
-        match daemon.handle(HuginnMindRequest::Whoami, now()) {
+        match daemon.handle(HuginnMindRequest::Whoami, now()).answered() {
             HuginnMindResponse::Whoami(MindStatus { documents, .. }) => documents,
             other => panic!("expected a status, got {other:?}"),
         }
@@ -539,14 +629,14 @@ mod tests {
             ("m-5b", request("m-5b", MIND_SERVICE_ID, "Admit", MIND_REQUEST_SCHEMA, admit), "operation-mismatch"),
         ];
         for (id, message, code) in cases {
-            let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), message, now());
+            let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), message, now()).reply();
             assert_eq!(rejected(&reply).0, id, "the client's own correlation key is echoed");
             assert_eq!(rejected(&reply).1, code);
         }
         // The sixth code, `not-an-operation-request`, is the codec's and is
         // pinned in `envelope::tests`: `answer` never produces it, because a
         // message of another family is answered with `Error` instead.
-        let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), CultNetMessage::Error { error: "hello".into(), code: None, details: None }, now());
+        let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), CultNetMessage::Error { error: "hello".into(), code: None, details: None }, now()).reply();
         let CultNetMessage::Error { error, .. } = &reply else { panic!("expected an error, got {reply:?}") };
         assert!(
             !error.is_empty() && error.contains("cultnet.operation_request.v0") && error.contains("cultnet.schema_catalog_request.v0"),
@@ -558,7 +648,7 @@ mod tests {
         // A mind's refusal is an answer on the response schema, not a failure
         // of the envelope: the client decodes it as the typed refusal it is.
         let read = HuginnMindRequest::Query { instance: slug(OTHER), selection: Selection::default(), semantic: None };
-        let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), encode_request("m-r", &read, None).unwrap(), now());
+        let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), encode_request("m-r", &read, None).unwrap(), now()).reply();
         let CultNetMessage::OperationResponse { status, payload_schema, .. } = &reply else {
             panic!("expected an operation response, got {reply:?}");
         };
@@ -576,7 +666,7 @@ mod tests {
             schema_ids: None,
             kinds: None,
         };
-        let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), catalog, now());
+        let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), catalog, now()).reply();
         let CultNetMessage::SchemaCatalogResponse { message_id, schemas } = reply else {
             panic!("expected a catalog response");
         };
@@ -749,7 +839,7 @@ mod tests {
         at: DateTime<Utc>,
     ) -> CultMeshCdnArtifactManifest {
         let registry = schema_registry().unwrap();
-        deferred_manifest(&answer(daemon, &registry, bodies, encode_request(message_id, read, None).unwrap(), at))
+        deferred_manifest(&answer(daemon, &registry, bodies, encode_request(message_id, read, None).unwrap(), at).reply())
     }
 
     /// Whether the first chunk of a manifest is still served at an injected time.
@@ -760,7 +850,7 @@ mod tests {
         at: DateTime<Utc>,
     ) -> bool {
         let registry = schema_registry().unwrap();
-        found(&answer(daemon, &registry, bodies, chunk_request(manifest, 0), at))
+        found(&answer(daemon, &registry, bodies, chunk_request(manifest, 0), at).reply())
     }
 
     /// An answer larger than one send can carry is deferred to the body plane
@@ -780,7 +870,7 @@ mod tests {
         let registry = schema_registry().unwrap();
 
         let sized = |daemon: &mut BodyDaemon, id: PipelineRef| {
-            let view = daemon.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id }, now());
+            let view = daemon.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id }, now()).answered();
             let message = encode_response("m-0", "view", &view, "huginn-yggdrasil").unwrap();
             encoded_len(&message)
         };
@@ -800,7 +890,7 @@ mod tests {
             ("m-v", HuginnMindRequest::View { instance: slug(INSTANCE), id: wide }),
         ];
         let expected: Vec<HuginnMindResponse> =
-            reads.iter().map(|(_, read)| daemon.handle(read.clone(), now())).collect();
+            reads.iter().map(|(_, read)| daemon.handle(read.clone(), now()).answered()).collect();
 
         let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap();
         let endpoint = format!("rudp://{}", hub.local_addr().unwrap());
@@ -1065,7 +1155,7 @@ mod tests {
         let registry = schema_registry().unwrap();
         let wide = spec_ref(&mut daemon, WIDE_CUT);
         let read = HuginnMindRequest::View { instance: slug(INSTANCE), id: wide };
-        let expected = rmp_serde::to_vec_named(&daemon.handle(read.clone(), now())).unwrap();
+        let expected = rmp_serde::to_vec_named(&daemon.handle(read.clone(), now()).answered()).unwrap();
         let t0 = now();
         let after = |seconds: i64| t0 + chrono::Duration::seconds(seconds);
         let mut bodies = fresh_bodies();
@@ -1073,7 +1163,7 @@ mod tests {
         let manifest = ask_deferred(&mut daemon, &mut bodies, "m-v", &read, t0);
         assert!(manifest.chunks.len() >= 2, "the fixture needs a second chunk to serve");
         let mut chunk = |index: usize, at: i64| {
-            answer(&mut daemon, &registry, &mut bodies, chunk_request(&manifest, index), after(at))
+            answer(&mut daemon, &registry, &mut bodies, chunk_request(&manifest, index), after(at)).reply()
         };
         assert!(found(&chunk(0, 59)), "inside the TTL");
         assert!(found(&chunk(1, 118)), "serving a chunk slid the expiry past the first TTL");
@@ -1086,7 +1176,7 @@ mod tests {
         let again = ask_deferred(&mut daemon, &mut bodies, "m-v2", &read, after(179));
         assert_eq!(again.content_hash, manifest.content_hash, "the same answer is the same body");
         let bytes = fetch_content(&again, MAX_DEFERRED_BODY_BYTES, |request| {
-            Ok(answer(&mut daemon, &registry, &mut bodies, request, after(180)))
+            Ok(answer(&mut daemon, &registry, &mut bodies, request, after(180)).reply())
         })
         .unwrap();
         assert_eq!(bytes, expected);
@@ -1102,7 +1192,7 @@ mod tests {
         let view = HuginnMindRequest::View { instance: slug(INSTANCE), id: wide };
         let query = document_query();
         let size = |daemon: &mut BodyDaemon, read: &HuginnMindRequest| {
-            rmp_serde::to_vec_named(&daemon.handle(read.clone(), now())).unwrap().len() as u64
+            rmp_serde::to_vec_named(&daemon.handle(read.clone(), now()).answered()).unwrap().len() as u64
         };
         let (view_bytes, query_bytes) = (size(&mut daemon, &view), size(&mut daemon, &query));
         // Room for the larger body alone, and not for both.
@@ -1186,7 +1276,7 @@ mod tests {
             ("record_key", chunk("m", "abc", &huge)),
         ];
         for (case, message) in refused {
-            let reply = answer(&mut daemon, &registry, &mut bodies, message, now());
+            let reply = answer(&mut daemon, &registry, &mut bodies, message, now()).reply();
             let CultNetMessage::Error { error, .. } = &reply else {
                 panic!("{case}: expected a transport error, got {reply:?}");
             };
@@ -1195,9 +1285,9 @@ mod tests {
         }
         assert_eq!(documents(&mut daemon), 0);
 
-        let reply = answer(&mut daemon, &registry, &mut bodies, whoami(&at, MIND_SERVICE_ID, "whoami", MIND_REQUEST_SCHEMA), now());
+        let reply = answer(&mut daemon, &registry, &mut bodies, whoami(&at, MIND_SERVICE_ID, "whoami", MIND_REQUEST_SCHEMA), now()).reply();
         assert_eq!(decode_response(&reply).unwrap().0, at, "a field of exactly the bound is served");
-        let reply = answer(&mut daemon, &registry, &mut bodies, chunk(&at, &at, &at), now());
+        let reply = answer(&mut daemon, &registry, &mut bodies, chunk(&at, &at, &at), now()).reply();
         assert!(!found(&reply), "an unknown chunk is answered found: false, not refused at the door");
     }
 
@@ -1233,7 +1323,7 @@ mod tests {
             ..Selection::default()
         };
         let HuginnMindResponse::Query(page) =
-            daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection, semantic: None }, now())
+            daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), selection, semantic: None }, now()).answered()
         else {
             panic!("expected a page");
         };

crates/huginn-daemon/tests/live_index_smoke.rs

@@ -11,7 +11,7 @@
 
 use std::collections::BTreeSet;
 
-use huginn_daemon::index::ollama::OllamaEmbedder;
+use huginn_daemon::index::ollama::{OllamaEmbedder, QUERY_INSTRUCTION};
 use huginn_daemon::index::qdrant::QdrantIndex;
 use huginn_daemon::index::{
     CollectionMeta, Described, Embedder, MANAGED_BY, Point, PointPayload, VectorIndex, point_id,
@@ -90,7 +90,17 @@ fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() {
             let (status, answer) = raw("POST", &format!("{qdrant}/collections/{name}/points/search"), Some(&query));
             assert_eq!(status, 200, "{answer}");
             assert!(answer.contains(&points[0].id), "{answer}");
-            println!("{name}: described, created, upserted {} points, queried", points.len());
+            // The read side: a paraphrase that shares no exact words with the
+            // first document finds it, through the adapter, with the query
+            // instruction the worker uses.
+            let asked = format!("Instruct: {QUERY_INSTRUCTION}
+Query: Who is responsible for keeping the memory?");
+            let query = embedder.embed(&[asked]).expect("the query embeds").remove(0);
+            let hits = store.search(name, &query, 2).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:?}");
+            println!("{name}: described, created, upserted {} points, queried, searched: {hits:?}", points.len());
         }
     }));
     for name in [&plain, &dotted] {

crates/huginn-mind/src/query.rs

@@ -16,12 +16,15 @@
 //! over the documents admitted at or before the page's `asOf`), the
 //! projection to a header or a whole view, and the typing of the edges.
 //!
-//! `semantic` is accepted by the type and refused typed until Cut 11b wires the
-//! read side of the index. No clock, no store handle, no network.
+//! `rank` is the semantic read: the daemon's index supplies candidate ids and
+//! 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 cultnet_rs::{Cursor, PROJECTION_DOCUMENT, Selection, select, validate};
+use cultnet_rs::{
+    Cursor, EdgeAnchor, EdgeMatch, Evaluation, LIMIT_MAX, LIMIT_MIN, PROJECTION_DOCUMENT, Row, Selection, select, validate,
+};
 use epiphany_pipeline::{
     ClaimOutcome, CommitRange, Date, FindingConfidence, FindingOrigin, FindingSeverity, Label, Line, OrgRepo,
     PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, ResolutionOutcome, RulingAuthority, Sha, Short,
@@ -72,8 +75,8 @@ pub struct PipelineDocumentView {
     pub status: PipelineStatus,
 }
 
-/// A semantic query, accepted by the type and refused until Cut 11 owns the
-/// index. Cut 11 also owns which fields of each kind are text, so no substring
+/// 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.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 pub struct SemanticQuery {
@@ -425,22 +428,12 @@ impl<S: MindStore> Mind<S> {
 
     /// One typed selection over the mind, as of one snapshot.
     ///
-    /// In order: `semantic` is refused before any read; the substrate checks
-    /// the selection's names; the organ's door checks its values; the
-    /// snapshot is the cursor's own `asOf` when there is a cursor and the
-    /// head otherwise (a cursor from beyond the head is one this mind could
-    /// not have minted); the substrate evaluates; the page is projected. The
-    /// evaluator's answer is never re-filtered here.
-    pub fn query(
-        &self,
-        selection: &Selection,
-        semantic: Option<&SemanticQuery>,
-    ) -> Result<PipelineSelectionPage, MindRefusal> {
-        if semantic.is_some() {
-            return Err(MindRefusal::Unavailable {
-                detail: "semantic query: the index is not wired (Cut 11)".into(),
-            });
-        }
+    /// In order: the substrate checks the selection's names; the organ's door
+    /// checks its values; the snapshot is the cursor's own `asOf` when there
+    /// is a cursor and the head otherwise (a cursor from beyond the head is
+    /// one this mind could not have minted); the substrate evaluates; the page
+    /// is projected. The evaluator's answer is never re-filtered here.
+    pub fn query(&self, selection: &Selection) -> Result<PipelineSelectionPage, MindRefusal> {
         validate(selection, &Vocabulary).map_err(cultnet_rs::SelectionRefusal::Invalid)?;
         let selection = refuse_values(selection)?;
         let head = receipt::head(self)?;
@@ -458,6 +451,118 @@ 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)
+    }
+
+    /// What a semantic query must satisfy before anything is embedded: the
+    /// selection passes the same two doors `query` does, carries no cursor
+    /// (a ranked answer is not pageable), and asks for between one and
+    /// `LIMIT_MAX` hits over text that says something. Returns the selection
+    /// as the door leaves it.
+    fn semantic_selection(selection: &Selection, semantic: &SemanticQuery) -> Result<Selection, MindRefusal> {
+        let refused = |field: &str, value: Option<String>, message: &str| MindRefusal::SelectionInvalid {
+            field: field.into(),
+            value,
+            message: message.into(),
+        };
+        if selection.cursor.as_deref().is_some_and(|cursor| !cursor.is_empty()) {
+            return Err(refused("cursor", None, "a semantic answer is ranked, not paged: it carries no cursor"));
+        }
+        if semantic.text.0.trim().is_empty() {
+            return Err(refused("semantic.text", None, "a semantic query needs text"));
+        }
+        if !(LIMIT_MIN..=LIMIT_MAX).contains(&semantic.top_k) {
+            return Err(refused(
+                "semantic.top_k",
+                Some(semantic.top_k.to_string()),
+                &format!("top_k is between {LIMIT_MIN} and {LIMIT_MAX}"),
+            ));
+        }
+        validate(selection, &Vocabulary).map_err(cultnet_rs::SelectionRefusal::Invalid)?;
+        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 page a semantic query answers, from the candidates the index gave.
+    ///
+    /// 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.
+    ///
+    /// 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. Only the best
+    /// `LIMIT_MAX` candidates are considered.
+    pub fn rank(
+        &self,
+        selection: &Selection,
+        semantic: &SemanticQuery,
+        hits: &[(PipelineRef, f32)],
+    ) -> Result<PipelineSelectionPage, MindRefusal> {
+        let selection = Self::semantic_selection(selection, semantic)?;
+        let head = receipt::head(self)?;
+        let rows = Reader::at(self, head)?.rows()?;
+
+        let mut best: Vec<&(PipelineRef, f32)> = hits.iter().collect();
+        best.sort_by(|a, b| b.1.total_cmp(&a.1));
+        best.truncate(LIMIT_MAX as usize);
+        let mut scores: BTreeMap<&str, f32> = BTreeMap::new();
+        for row in &rows {
+            if let Some((_, score)) = best.iter().find(|(hit, _)| hit.kind == row.kind && hit.id.0 == row.key) {
+                scores.insert(row.key.as_str(), *score);
+            }
+        }
+        let mut keys: Vec<String> = scores.keys().map(|key| key.to_string()).collect();
+        if let Some(allowed) = &selection.keys {
+            keys.retain(|key| allowed.contains(key));
+        }
+        let limit = selection.limit.unwrap_or(LIMIT_MAX).clamp(LIMIT_MIN, LIMIT_MAX).min(semantic.top_k) as usize;
+
+        let mut evaluation = if keys.is_empty() {
+            Evaluation { rows: Vec::new(), matched: 0, edges: Vec::new(), next_cursor: None }
+        } else {
+            let narrowed = Selection { keys: Some(keys), limit: Some(LIMIT_MAX), cursor: None, ..selection.clone() };
+            select(&Vocabulary, &rows, &narrowed, head, 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 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)
+            .into_iter()
+            .filter_map(|edge| {
+                let owner = match edge.anchor {
+                    EdgeAnchor::Citee => &edge.to,
+                    EdgeAnchor::Citer => &edge.from,
+                };
+                position(owner.kind, &owner.key).map(|at| (at, edge))
+            })
+            .collect();
+        kept.sort_by_key(|(at, _)| *at);
+        evaluation.edges = kept.into_iter().map(|(_, edge)| edge).collect();
+        Self::page(&selection, evaluation, head)
+    }
+
+    /// The projection of an evaluation: a header or a whole view per row, and
+    /// the typed edges a hop traversed.
+    fn page(
+        selection: &Selection,
+        evaluation: Evaluation<SelectionRow>,
+        as_of: u64,
+    ) -> Result<PipelineSelectionPage, MindRefusal> {
         let items = if selection.projection == PROJECTION_DOCUMENT {
             PipelinePageItems::Documents(evaluation.rows.iter().map(|row| row.view.clone()).collect())
         } else {
@@ -534,7 +639,7 @@ mod tests {
 
     /// The refusal a selection earns from a mind that would otherwise answer.
     fn refused(mind: &Mind<MemoryStore>, selection: &Selection) -> MindRefusal {
-        mind.query(selection, None).expect_err("the selection is refused")
+        mind.query(selection).expect_err("the selection is refused")
     }
 
     fn invalid(field: &str, value: &str) -> impl Fn(&MindRefusal) -> bool {
@@ -687,7 +792,7 @@ mod tests {
         let refuses = |store: MemoryStore, label: &str| {
             let broken_mind = opened(store, INSTANCE);
             assert!(
-                matches!(broken_mind.query(&Selection::default(), None), Err(MindRefusal::Unavailable { .. })),
+                matches!(broken_mind.query(&Selection::default()), Err(MindRefusal::Unavailable { .. })),
                 "{label}: query must refuse an undense chain"
             );
             assert!(
@@ -743,7 +848,7 @@ mod tests {
             orphaned.view(&r(K::Question, &orphan)).err(),
             Some(MindRefusal::Unavailable { detail: detail.clone() })
         );
-        assert_eq!(orphaned.query(&Selection::default(), None).err(), Some(MindRefusal::Unavailable { detail }));
+        assert_eq!(orphaned.query(&Selection::default()).err(), Some(MindRefusal::Unavailable { detail }));
 
         // The other half of the same rule: A10 admits one write per identity,
         // so two receipts naming one document is a store this organ cannot
@@ -763,7 +868,7 @@ mod tests {
         .unwrap();
         doubled.plant(schema_cache().unwrap().prepare_entry_named(&forged.receipt_id, &forged).unwrap().0);
         assert_eq!(
-            opened(doubled, INSTANCE).query(&Selection::default(), None).err(),
+            opened(doubled, INSTANCE).query(&Selection::default()).err(),
             Some(MindRefusal::Unavailable {
                 detail: format!("document {}/{q1} is written by two receipts", K::Question.type_id()),
             })
@@ -795,11 +900,11 @@ mod tests {
             }
             opened(copy, INSTANCE)
         };
-        let expected = mind.query(&Selection::default(), None).unwrap();
+        let expected = mind.query(&Selection::default()).unwrap();
         let constant = vec!["2000-01-01T00:00:00Z".to_string(); rows.len()];
         let reversed = rows.iter().rev().map(|row| row.stored_at.clone()).collect::<Vec<_>>();
         for other in [reopened(constant), reopened(reversed)] {
-            assert_eq!(other.query(&Selection::default(), None).unwrap(), expected);
+            assert_eq!(other.query(&Selection::default()).unwrap(), expected);
         }
     }
 
@@ -841,7 +946,7 @@ mod tests {
     }
 
     fn matching(mind: &Mind<MemoryStore>, selection: Selection) -> Vec<String> {
-        let mut matched = ids(&mind.query(&selection, None).unwrap());
+        let mut matched = ids(&mind.query(&selection).unwrap());
         matched.sort();
         matched
     }
@@ -938,11 +1043,11 @@ mod tests {
         committed(admit(&mut mind, vec![question_n(9)]));
         committed(admit(&mut mind, vec![question_n(1)]));
         let selection = of_kinds(&[K::Question]);
-        assert_eq!(ids(&mind.query(&selection, None).unwrap()), vec![id("question", "Q9"), id("question", "Q1")]);
+        assert_eq!(ids(&mind.query(&selection).unwrap()), vec![id("question", "Q9"), id("question", "Q1")]);
         let newest_first = Selection { descending: true, ..selection };
-        assert_eq!(ids(&mind.query(&newest_first, None).unwrap()), vec![id("question", "Q1"), id("question", "Q9")]);
+        assert_eq!(ids(&mind.query(&newest_first).unwrap()), vec![id("question", "Q1"), id("question", "Q9")]);
         // The facts on a header name the ordinal the order came from.
-        let page = mind.query(&of_kinds(&[K::Question]), None).unwrap();
+        let page = mind.query(&of_kinds(&[K::Question])).unwrap();
         let ordinals: Vec<u64> = headers(&page).iter().map(|header| header.admission.ordinal).collect();
         assert_eq!(ordinals, vec![2, 3]);
     }
@@ -959,13 +1064,13 @@ mod tests {
         }
         let questions = of_kinds(&[K::Question]);
         for (limit, expected) in [(None, 200), (Some(500), 200), (Some(0), 1), (Some(5), 5)] {
-            let page = mind.query(&Selection { limit, ..questions.clone() }, None).unwrap();
+            let page = mind.query(&Selection { limit, ..questions.clone() }).unwrap();
             assert_eq!(ids(&page).len(), expected, "limit {limit:?}");
             assert_eq!(page.matched, 201, "the count is taken before the cap");
         }
-        let first = mind.query(&questions, None).unwrap();
+        let first = mind.query(&questions).unwrap();
         assert_eq!(ids(&first)[0], id("question", "Q10"), "the earliest batch first, and inside it key order");
-        let second = mind.query(&Selection { cursor: first.next.clone(), ..questions }, None).unwrap();
+        let second = mind.query(&Selection { cursor: first.next.clone(), ..questions }).unwrap();
         assert_eq!(ids(&second), vec![id("question", "Q201")], "the last batch's last key is the one the cap dropped");
         assert_eq!((second.matched, second.next), (201, None));
         assert_eq!(second.as_of, first.as_of);
@@ -984,7 +1089,7 @@ mod tests {
         // Newest first, so page 1's last row is the head row: the boundary
         // sits at exactly `asOf`.
         let walk = Selection { descending: true, limit: Some(2), ..of_kinds(&[K::Question]) };
-        let first = mind.query(&walk, None).unwrap();
+        let first = mind.query(&walk).unwrap();
         assert_eq!(ids(&first), vec![id("question", "Q3"), id("question", "Q2")]);
         assert_eq!((first.matched, first.as_of), (3, 4));
         let cursor = first.next.clone().expect("Q1 is left");
@@ -992,16 +1097,16 @@ mod tests {
         let q1 = id("question", "Q1");
         committed(admit(&mut mind, vec![answering("R1", &q1), question_n(4)]));
 
-        let second = mind.query(&Selection { cursor: Some(cursor.clone()), ..walk.clone() }, None).unwrap();
+        let second = mind.query(&Selection { cursor: Some(cursor.clone()), ..walk.clone() }).unwrap();
         assert_eq!(ids(&second), vec![q1.clone()]);
         assert_eq!(headers(&second)[0].status, PipelineStatusSummary::InForce, "the closing ruling landed after asOf");
         assert_eq!((second.matched, second.as_of, second.next.as_deref()), (3, 4, None));
 
         // A fresh walk starts at the new head and sees all of it.
-        let fresh = mind.query(&walk, None).unwrap();
+        let fresh = mind.query(&walk).unwrap();
         assert_eq!((fresh.matched, fresh.as_of), (4, 5));
         assert_eq!(ids(&fresh), vec![id("question", "Q4"), id("question", "Q3")]);
-        let closed = mind.query(&Selection { cursor: fresh.next.clone(), ..walk.clone() }, None).unwrap();
+        let closed = mind.query(&Selection { cursor: fresh.next.clone(), ..walk.clone() }).unwrap();
         assert!(matches!(headers(&closed)[1].status, PipelineStatusSummary::Resolved { .. }), "Q1 is closed at head");
 
         // A cursor is bound to its selection: change the order and it is not
@@ -1023,10 +1128,10 @@ mod tests {
             cursor: Some(Cursor::mint(as_of, &rows[0], &selection, mind.cursor_key())),
             ..selection.clone()
         };
-        assert!(mind.query(&at(head), None).is_ok(), "a cursor at exactly the head answers");
+        assert!(mind.query(&at(head)).is_ok(), "a cursor at exactly the head answers");
         assert!(matches!(refused(&mind, &at(head + 1)), MindRefusal::CursorInvalid { .. }));
         let stale = at(1);
-        assert!(mind.query(&stale, None).is_ok(), "an older snapshot is still answerable: the mind is append-only");
+        assert!(mind.query(&stale).is_ok(), "an older snapshot is still answerable: the mind is append-only");
     }
 
     /// The door refuses a value outside its alias's domain, naming the field
@@ -1034,7 +1139,7 @@ mod tests {
     #[test]
     fn a_root_outside_the_grammar_is_refused_not_empty() {
         let mind = worked();
-        assert!(mind.query(&with(Selection::default(), any_of("root", &[CAMPAIGN])), None).unwrap().matched > 0);
+        assert!(mind.query(&with(Selection::default(), any_of("root", &[CAMPAIGN]))).unwrap().matched > 0);
         // The bad value is the second in its list and the predicate is the
         // second in its selection: the door reads them all.
         let bad = with(with(Selection::default(), any_of("in_force", &["true"])), any_of("root", &[CAMPAIGN, "a/../b"]));
@@ -1060,8 +1165,8 @@ mod tests {
         let mind = worked();
         let severity = |value: &str| with(of_kinds(&[K::Finding]), any_of("severity", &[value]));
         assert!(invalid("fields[0].values", "high")(&refused(&mind, &severity("high"))));
-        assert_eq!(ids(&mind.query(&severity("High"), None).unwrap()), vec![id("finding", "cut-9.s1.F1")]);
-        assert!(mind.query(&severity("Low"), None).unwrap().matched == 0, "a value in the domain that matches nothing is empty");
+        assert_eq!(ids(&mind.query(&severity("High")).unwrap()), vec![id("finding", "cut-9.s1.F1")]);
+        assert!(mind.query(&severity("Low")).unwrap().matched == 0, "a value in the domain that matches nothing is empty");
         let in_force = with(Selection::default(), any_of("in_force", &["yes"]));
         assert!(invalid("fields[0].values", "yes")(&refused(&mind, &in_force)));
     }
@@ -1077,7 +1182,7 @@ mod tests {
             for alias in ["root", "in_force", "faculty", "repo", "cut", "severity", "confidence", "origin", "authority", "claim_outcome", "outcome"] {
                 for value in cultnet_rs::Row::values(row, alias) {
                     let selection = with(Selection::default(), any_of(alias, &[&value]));
-                    assert!(mind.query(&selection, None).unwrap().matched >= 1, "{alias}={value}");
+                    assert!(mind.query(&selection).unwrap().matched >= 1, "{alias}={value}");
                 }
             }
         }
@@ -1097,7 +1202,7 @@ mod tests {
         }
         let q1 = id("question", "Q1");
         let held = Selection { keys: Some(vec![q1.clone()]), ..Selection::default() };
-        assert_eq!(ids(&mind.query(&held, None).unwrap()), vec![q1]);
+        assert_eq!(ids(&mind.query(&held).unwrap()), vec![q1]);
     }
 
     /// V24 at the selection's door: a `cites` target whose kind and key
@@ -1109,7 +1214,7 @@ mod tests {
         let wrong = Selection { cites: Some(cites(K::Ruling, &q1, None)), ..Selection::default() };
         assert!(invalid("cites.target", &q1)(&refused(&mind, &wrong)));
         let right = Selection { cites: Some(cites(K::Question, &q1, Some("source"))), ..Selection::default() };
-        assert_eq!(ids(&mind.query(&right, None).unwrap()), vec![id("follow_up", "FU-1")]);
+        assert_eq!(ids(&mind.query(&right).unwrap()), vec![id("follow_up", "FU-1")]);
     }
 
     /// The substrate's own refusals cross as themselves: an undeclared alias
@@ -1133,7 +1238,7 @@ mod tests {
         committed(admit(&mut mind, vec![question_n(1)]));
         let q1 = id("question", "Q1");
         let one = Selection { keys: Some(vec![q1.clone()]), ..Selection::default() };
-        let page = mind.query(&one, None).unwrap();
+        let page = mind.query(&one).unwrap();
         let expected = view(&mind, K::Question, &q1);
         assert_eq!(page.edges, None, "no hop, no edges");
         assert_eq!(headers(&page), [PipelineDocumentSummary::of(&expected)]);
@@ -1145,7 +1250,7 @@ mod tests {
             raised_in: None,
             asked_on: date(),
         });
-        let whole = mind.query(&Selection { projection: "document".into(), ..one }, None).unwrap();
+        let whole = mind.query(&Selection { projection: "document".into(), ..one }).unwrap();
         assert_eq!(whole.items, PipelinePageItems::Documents(vec![expected]));
     }
 
@@ -1221,7 +1326,7 @@ mod tests {
         for role in CitationRole::ALL {
             let selection =
                 Selection { cited: Some(Incoming { role: role.name().into(), exists: true }), ..Selection::default() };
-            let page = mind.query(&selection, None).unwrap();
+            let page = mind.query(&selection).unwrap();
             let edges = page.edges.expect("a hop carries edges");
             let mut found: Vec<(CitationRole, String, String)> =
                 edges.iter().map(|edge| (edge.role, edge.from.id.0.clone(), edge.to.id.0.clone())).collect();
@@ -1258,7 +1363,7 @@ mod tests {
             }
         }
         let history = Selection { cites: Some(cites(K::Question, &q1, Some("subject"))), ..of_kinds(&[K::Resolution]) };
-        let page = mind.query(&history, None).unwrap();
+        let page = mind.query(&history).unwrap();
         let expected: Vec<String> = (1..=10).map(|cycle| id("resolution", &format!("question.Q1.n{cycle}"))).collect();
         assert_eq!(ids(&page), expected, "ordinal order, so n10 follows n9");
         let statuses: Vec<bool> =
@@ -1272,7 +1377,7 @@ mod tests {
         // resolutions by `subject`, the ten rulings by `answers` and the
         // follow-up by `source`.
         let unroled = Selection { cites: Some(cites(K::Question, &q1, None)), ..Selection::default() };
-        let page = mind.query(&unroled, None).unwrap();
+        let page = mind.query(&unroled).unwrap();
         assert_eq!(page.matched, 21);
         assert!(page.edges.unwrap().iter().any(|edge| edge.role == CitationRole::Answers));
     }
@@ -1285,7 +1390,7 @@ mod tests {
         committed(admit(&mut mind, vec![hand_off(INSTANCE, OTHER_INSTANCE, REPO, &[])]));
         let assignments = |root: &str, repo_name: &str| {
             let selection = with(with(of_kinds(&[K::Stewardship]), any_of("root", &[root])), any_of("repo", &[repo_name]));
-            mind.query(&selection, None).unwrap()
+            mind.query(&selection).unwrap()
         };
         let mine = assignments(INSTANCE, REPO);
         assert_eq!(ids(&mine), vec![format!("{INSTANCE}:stewardship:gamecult_-epiphany.n1")]);
@@ -1342,7 +1447,7 @@ mod tests {
             cited: Some(Incoming { role: role.into(), exists: false }),
             ..selection
         };
-        let open = |selection: Selection| ids(&mind.query(&selection, None).unwrap());
+        let open = |selection: Selection| ids(&mind.query(&selection).unwrap());
 
         assert_eq!(open(standing(mine(&[K::Question]))), vec![q1.clone()]);
         assert_eq!(open(standing(mine(&[K::Finding]))), vec![f1]);
@@ -1357,54 +1462,168 @@ mod tests {
         assert_eq!(open(standing(of_kinds(&[K::Question]))).len(), 2);
     }
 
-    /// `semantic` is refused typed, before the image or the receipts are
-    /// read, and never answered with an empty page that reads like "nothing
-    /// matched". Where the check sits is the rule: below the reader, a store
-    /// the reader refuses would answer the integrity fault instead, which is
-    /// not what the caller asked about.
-    #[test]
-    fn semantic_query_refuses_typed_until_wired() {
-        let store = MemoryStore::new();
-        let mut mind = opened(store.clone(), INSTANCE);
-        seed(&mut mind);
-        let q1 = id("question", "Q1");
-        committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A")]));
-        let selection = with(with(of_kinds(&[K::Question]), any_of("root", &[CAMPAIGN])), any_of("in_force", &["true"]));
-        let semantic = SemanticQuery { text: "what did we rule about keys".into(), top_k: 5 };
-        let unwired = MindRefusal::Unavailable { detail: "semantic query: the index is not wired (Cut 11)".into() };
-        assert_eq!(mind.query(&selection, Some(&semantic)).err(), Some(unwired.clone()));
-        assert!(mind.query(&selection, None).is_ok());
-        // A selection that is itself invalid still earns the semantic
-        // refusal first: nothing about it is read.
-        let malformed = with(Selection::default(), any_of("root", &["a/../b"]));
-        assert_eq!(mind.query(&malformed, Some(&semantic)).err(), Some(unwired.clone()));
+    fn hit(kind: K, key: &str, score: f32) -> (PipelineRef, f32) {
+        (r(kind, key), score)
+    }
 
-        // Two receipts naming one write is a store the reader refuses to build
-        // over at all. The semantic refusal still lands, because it is decided
-        // before the image and the receipts are read.
-        let doubled = MemoryStore::new();
-        for row in store.rows() {
-            doubled.plant(row);
+    fn asking(top_k: u32) -> SemanticQuery {
+        SemanticQuery { text: "who owns the state".into(), top_k }
+    }
+
+    fn standing(selection: Selection) -> Selection {
+        with(selection, any_of("in_force", &["true"]))
+    }
+
+    /// 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.
+    struct Shelf {
+        mind: Mind<MemoryStore>,
+        q1: String,
+        q2: String,
+        withdrawn_resolution: String,
+        r3: String,
+        r4: String,
+    }
+
+    fn shelf() -> Shelf {
+        let mut mind = seeded();
+        let (q1, q2) = (id("question", "Q1"), id("question", "Q2"));
+        committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A"), question("Q2", &["A", "B"], "A")]));
+        committed(admit(&mut mind, vec![answering("R1", &q1)]));
+        committed(admit(&mut mind, vec![answering("R2", &q2)]));
+        let withdrawn_resolution = id("resolution", "question.Q2.n1");
+        committed(admit(&mut mind, vec![resolution(r(K::Resolution, &withdrawn_resolution), withdrawn())]));
+        let (r3, r4) = (id("ruling", "R3"), id("ruling", "R4"));
+        committed(admit(&mut mind, vec![D::Ruling(ruling("R3")), D::Ruling(ruling("R4"))]));
+        committed(admit(&mut mind, vec![resolution(r(K::Ruling, &r3), superseded(&[r(K::Ruling, &r4)]))]));
+        Shelf { mind, q1, q2, withdrawn_resolution, r3, r4 }
+    }
+
+    /// The index never deletes and carries no status, so a hit is a candidate
+    /// and nothing more. Here it names a resolved question, a withdrawn
+    /// resolution, a superseded ruling, a document the mind does not hold and
+    /// a real id under the wrong kind, all scoring above the two documents
+    /// that stand; the standing ones are the whole answer.
+    #[test]
+    fn a_hit_is_judged_by_the_mind_and_never_by_the_index() {
+        let shelf = shelf();
+        let hits = [
+            hit(K::Question, &shelf.q1, 0.9),
+            hit(K::Resolution, &shelf.withdrawn_resolution, 0.8),
+            hit(K::Ruling, &shelf.r3, 0.7),
+            hit(K::Question, &id("question", "Q99"), 0.6),
+            hit(K::Question, &shelf.r4, 0.55),
+            hit(K::Ruling, &shelf.r4, 0.5),
+            hit(K::Question, &shelf.q2, 0.4),
+        ];
+        let everything = shelf.mind.rank(&Selection::default(), &asking(10), &hits).unwrap();
+        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);
+
+        let page = shelf.mind.rank(&standing(Selection::default()), &asking(10), &hits).unwrap();
+        assert_eq!(ids(&page), vec![shelf.r4.clone(), shelf.q2.clone()]);
+        assert_eq!((page.matched, page.next), (2, 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();
+        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().matched, 0);
+    }
+
+    /// The order is the score's, highest first, and ordinal only settles a
+    /// tie; the page is cut to the selection's limit and to top_k, and says
+    /// how many matched.
+    #[test]
+    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()), 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()), 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();
+        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();
+        assert_eq!((ids(&page), page.matched), (vec![shelf.r4.clone()], 2), "top_k bounds the page too");
+    }
+
+    /// Membership is the selection's alone: a kind, a key allowlist and a
+    /// citation predicate each cut the candidates exactly as they cut a plain
+    /// query, and the citation predicate sees the whole mind, not only the
+    /// candidates (Q2 is named by a resolution that is not a hit).
+    #[test]
+    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()), 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()), 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().matched, 0);
+        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()), 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()), vec![shelf.r4.clone()]);
+    }
+
+    /// A citation hop is answered over the whole mind, and its edges follow the
+    /// ranked page: after the cut to the limit only the edges anchored on a kept
+    /// row remain, in the page's order and not the evaluator's. An empty cursor
+    /// is no cursor.
+    #[test]
+    fn a_hop_keeps_only_the_edges_of_the_ranked_page_in_its_order() {
+        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();
+        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();
+        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");
+        assert_eq!(kept[0].to.id.0, shelf.q2);
+        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();
+        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.
+    #[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)
         }
-        let forged = crate::receipt::candidate(
-            &slug(INSTANCE),
-            provenance(Faculty::Soul),
-            &[],
-            &[prepare(&question("Q1", &["A", "B"], "A")), prepare(&question("Q8", &["A", "B"], "A"))],
-            3, // head + 1: dense, so this plants only the duplicate identity, not an F1 density fault too.
-            now(),
-        )
-        .unwrap();
-        doubled.plant(schema_cache().unwrap().prepare_entry_named(&forged.receipt_id, &forged).unwrap().0);
-        let unreadable = opened(doubled, INSTANCE);
-        assert_eq!(
-            unreadable.query(&Selection::default(), None).err(),
-            Some(MindRefusal::Unavailable {
-                detail: format!("document {}/{q1} is written by two receipts", K::Question.type_id()),
-            }),
-            "the reader refuses this store"
-        );
-        assert_eq!(unreadable.query(&selection, Some(&semantic)).err(), Some(unwired), "and the semantic refusal precedes reading it");
+        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:?}");
+        }
+        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 blank = SemanticQuery { text: "  ".into(), top_k: 3 };
+        assert!(on("semantic.text", &shelf.mind.rank(&Selection::default(), &blank, &hits).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());
     }
 
     /// A summary is bounded by a named constant whatever the document's lists
@@ -1452,7 +1671,7 @@ mod tests {
         };
         committed(mind.admit(batch, now()));
 
-        let page = mind.query(&Selection::default(), None).unwrap();
+        let page = mind.query(&Selection::default()).unwrap();
         let sizes: Vec<usize> =
             headers(&page).iter().map(|summary| rmp_serde::to_vec_named(summary).unwrap().len()).collect();
         let widest = *sizes.iter().max().unwrap();

schemas/cultnet/huginn.mind_request.v1.schema.json

@@ -1678,7 +1678,7 @@
       ]
     },
     "SemanticQuery": {
-      "description": "A semantic query, accepted by the type and refused until Cut 11 owns the\nindex. Cut 11 also owns which fields of each kind are 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.",
       "type": "object",
       "properties": {
         "text": {