Merge Cut 11a: the index write side

huginn-mind::index defines each kind's search text (pure). The daemon gains
Ollama and Qdrant adapters (ureq 3, no TLS) behind Embedder/VectorIndex
ports, and one worker thread whose Projector owns reconcile. Admission
never waits on the index. Reconcile is the only judge of missing points,
and it runs again after any failure. The collection is labelled with the
mind's genesis receipt id and keyed on model identity (name, digest,
dims, text version), never the URL. whoami reports IndexStatus. The live
smoke passed against Qdrant on Yggdrasil and Raven's Ollama
(qwen3-embedding:0.6b, 1024 dims).

Soul closed it after one fix batch. For Cut 11b: every search hit is
joined back through the mind and filtered by in-force; model identity is
re-checked per flush. Noted limit: Ollama truncates at 4096 tokens.
mind 96, daemon 47.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 10:31:03 +02:00
commit efaeebffc0
21 changed files with 2658 additions and 89 deletions

64
Cargo.lock generated
View file

@ -91,6 +91,12 @@ version = "3.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
[[package]]
name = "bytes"
version = "1.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
[[package]]
name = "cc"
version = "1.4.6"
@ -445,6 +451,22 @@ dependencies = [
"digest",
]
[[package]]
name = "http"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0"
dependencies = [
"bytes",
"itoa",
]
[[package]]
name = "httparse"
version = "1.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
[[package]]
name = "huginn-daemon"
version = "0.1.0"
@ -457,8 +479,11 @@ dependencies = [
"rmp-serde",
"schemars",
"serde",
"serde_json",
"sha2",
"signal-hook",
"tempfile",
"ureq",
]
[[package]]
@ -573,6 +598,12 @@ version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
[[package]]
name = "percent-encoding"
version = "2.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
[[package]]
name = "pin-project-lite"
version = "0.2.17"
@ -967,7 +998,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
dependencies = [
"fastrand",
"getrandom 0.3.4",
"getrandom 0.4.3",
"once_cell",
"rustix",
"windows-sys 0.61.2",
@ -995,6 +1026,37 @@ dependencies = [
"subtle",
]
[[package]]
name = "ureq"
version = "3.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0"
dependencies = [
"base64",
"log",
"percent-encoding",
"ureq-proto",
"utf8-zero",
]
[[package]]
name = "ureq-proto"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c"
dependencies = [
"base64",
"http",
"httparse",
"log",
]
[[package]]
name = "utf8-zero"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e"
[[package]]
name = "uuid"
version = "1.26.1"

View file

@ -17,9 +17,15 @@ Upstream: `https://github.com/GameCult/Huginn.git`
mind over CultNet RUDP: admission and the read side ride
`cultnet.operation_request.v0` and `cultnet.operation_response.v0`, and the
two wire schemas answer a schema catalog request.
- Retrieval is to depend on Qdrant directly, refusing loudly when Qdrant is
unreachable rather than falling back to a second store or a second writer.
No crate opens that connection yet.
- Retrieval depends on Qdrant directly, with no second store and no second
writer. `huginn-daemon` keeps one collection per mind, `huginn_mind_<instance>`,
as a projection of the mind: one worker thread embeds each indexable
document through Ollama and writes its point, off the serving path, and a
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.
- Mind state is not version-controlled. It lives in Huginn's store, not in
any repository.

View file

@ -26,7 +26,15 @@ huginn-mind = { path = "../huginn-mind" }
rmp-serde = "1"
schemars = "1"
serde = { version = "1", features = ["derive"] }
# The vector store's and the embedder's HTTP APIs are the xenos boundary, and
# JSON exists in `index/ollama.rs` and `index/qdrant.rs` only. `ureq` 3 with
# default features off is plain HTTP with no TLS (`rustls` and `gzip` are its
# only defaults), which is all the private mesh needs. It stays below 3.4,
# whose `base64` 0.23 would put a second `base64` in a tree that has one.
serde_json = "1"
sha2 = "0.10"
signal-hook = "0.3"
ureq = { version = "~3.3", default-features = false }
[dev-dependencies]
tempfile = "3"

View file

@ -2,28 +2,25 @@
//! rule. `now` is the caller's, so a test pins it and the loop is the only
//! thing that reads a wall clock.
use std::path::Path;
use chrono::{DateTime, Utc};
use huginn_mind::epiphany_pipeline::{PipelineRef, Slug};
use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse};
use huginn_mind::{Mind, MindRefusal, MindStore, OwnedRedbMessagePackBackingStore, PipelineAdmissionOutcome};
use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, IndexStatus};
use huginn_mind::{Mind, MindStore, PipelineAdmissionOutcome};
/// Where Cut 11's index attaches. Called after a batch commits, with the refs
/// the batch landed, derived writes included; the sink may read each through
/// 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.
/// 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.
pub trait IndexSink<S: MindStore> {
fn committed(&mut self, mind: &Mind<S>, writes: &[PipelineRef]) -> anyhow::Result<()>;
fn status(&self) -> IndexStatus;
}
/// This cut's index: there is none.
pub struct NoIndex;
impl<S: MindStore> IndexSink<S> for NoIndex {
fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
Ok(())
}
/// The envelope's `source_runtime_id` on every response a daemon serving this
/// instance sends. Also what the socket is bound under, before a daemon exists.
pub fn runtime_id(instance: &Slug) -> String {
format!("huginn-{}", instance.0)
}
/// One mind and whatever indexes what lands in it.
@ -32,17 +29,14 @@ pub struct Daemon<S: MindStore, I: IndexSink<S>> {
index: I,
}
impl Daemon<OwnedRedbMessagePackBackingStore, NoIndex> {
/// Opens the instance's mind, taking the store's exclusive lock for the
/// daemon's lifetime. Ruling 15: a mind that will not open is a refusal
/// here, before any socket exists; this function takes no address and
/// cannot bind one.
pub fn open(state_root: &Path, instance: &Slug) -> Result<Self, MindRefusal> {
Ok(Self { mind: Mind::open(state_root, instance)?, index: NoIndex })
}
}
impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
/// A daemon over a mind that is already open. `Mind::open` takes the
/// store's exclusive lock for the daemon's lifetime, and ruling 15 has it
/// refuse before any socket exists: the caller opens the mind first.
pub fn new(mind: Mind<S>, index: I) -> Self {
Self { mind, index }
}
pub fn with_index<J: IndexSink<S>>(self, index: J) -> Daemon<S, J> {
Daemon { mind: self.mind, index }
}
@ -51,9 +45,13 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
&self.mind
}
pub fn index(&self) -> &I {
&self.index
}
/// The envelope's `source_runtime_id` on every response this daemon sends.
pub fn runtime_id(&self) -> String {
format!("huginn-{}", self.mind.instance().0)
runtime_id(self.mind.instance())
}
/// One request, one answer. `Admit` goes to the mind whole, because the
@ -70,7 +68,7 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
return HuginnMindResponse::Refused(refusal);
}
match request {
HuginnMindRequest::Whoami => HuginnMindResponse::Whoami(self.mind.status()),
HuginnMindRequest::Whoami => HuginnMindResponse::Whoami(self.mind.status(self.index.status())),
HuginnMindRequest::Admit(batch) => {
let outcome = self.mind.admit(batch, now);
if let PipelineAdmissionOutcome::Committed { writes, .. } = &outcome
@ -104,6 +102,8 @@ pub(crate) mod tests {
StructuralDelta, Title, VerificationTest,
};
use huginn_mind::wire::MindStatus;
use huginn_mind::{MindRefusal, OwnedRedbMessagePackBackingStore};
use std::path::Path;
use cultnet_rs::Selection;
use huginn_mind::{Faculty, PipelineAdmissionBatch, PipelineProvenance, PipelinePageItems, PipelineStatus, SemanticQuery};
use std::sync::{Arc, Mutex};
@ -295,7 +295,7 @@ pub(crate) mod tests {
/// real-sized answer either way.
pub(crate) fn seeded_wide() -> (TempDir, Daemon<OwnedRedbMessagePackBackingStore, NoIndex>) {
let root = tempfile::tempdir().unwrap();
let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
for documents in [
campaign_seed(),
vec![cut_spec(WIDE_CUT, WIDE_CHANGES), cut_spec(FITTING_CUT, FITTING_CHANGES)],
@ -314,7 +314,7 @@ pub(crate) mod tests {
/// removes the store.
pub(crate) fn seeded() -> (TempDir, Daemon<OwnedRedbMessagePackBackingStore, NoIndex>) {
let root = tempfile::tempdir().unwrap();
let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
assert!(
matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })),
@ -323,6 +323,27 @@ pub(crate) mod tests {
(root, daemon)
}
/// No index: for the tests whose subject is not the projection.
pub(crate) struct NoIndex;
impl<S: MindStore> IndexSink<S> for NoIndex {
fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
Ok(())
}
fn status(&self) -> IndexStatus {
IndexStatus::Current
}
}
/// A daemon over the instance's mind under `state_root`, with no index.
pub(crate) fn open_unindexed(
state_root: &Path,
instance: &Slug,
) -> Result<Daemon<OwnedRedbMessagePackBackingStore, NoIndex>, MindRefusal> {
Ok(Daemon::new(Mind::open(state_root, instance)?, NoIndex))
}
#[derive(Clone, Default)]
struct RecordingIndex {
seen: Arc<Mutex<Vec<(Vec<String>, Vec<bool>)>>>,
@ -335,6 +356,10 @@ pub(crate) mod tests {
self.seen.lock().unwrap().push((ids, readable));
Ok(())
}
fn status(&self) -> IndexStatus {
IndexStatus::Current
}
}
struct FailingIndex;
@ -343,6 +368,10 @@ pub(crate) mod tests {
fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
Err(anyhow::anyhow!("the index is not there"))
}
fn status(&self) -> IndexStatus {
IndexStatus::Current
}
}
fn status(daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, impl IndexSink<OwnedRedbMessagePackBackingStore>>) -> MindStatus {
@ -357,7 +386,7 @@ pub(crate) mod tests {
#[test]
fn a_batch_round_trips_typed_through_handle_without_a_socket() {
let root = tempfile::tempdir().unwrap();
let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = outcome else {
panic!("expected a commit");
@ -372,6 +401,7 @@ pub(crate) mod tests {
schema_epoch: "epiphany.pipeline.epoch.v2".into(),
documents: 1,
receipts: 1,
index: IndexStatus::Current,
}
);
assert_eq!(daemon.runtime_id(), "huginn-yggdrasil");
@ -406,7 +436,7 @@ pub(crate) mod tests {
#[test]
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 = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
let landed = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = landed else {
panic!("expected a commit");
@ -626,7 +656,7 @@ pub(crate) mod tests {
let (question, ruling) = question_and_ruling();
let mut daemon = daemon.with_index(FailingIndex);
let root = tempfile::tempdir().unwrap();
let mut fresh = Daemon::open(root.path(), &slug(INSTANCE)).unwrap().with_index(FailingIndex);
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());
let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = outcome else {
@ -639,28 +669,28 @@ pub(crate) mod tests {
assert_eq!(status(&mut daemon).documents, 4, "identity, question, ruling and the derived resolution");
}
/// Ruling 15: a mind that will not open is refused, and `Daemon::open`
/// takes no address, so no socket can precede it. `startup`'s order is
/// Ruling 15: a mind that will not open is refused, and opening it takes
/// no address, so no socket can precede it. `startup`'s order is
/// pinned in `serve::tests`.
#[test]
fn the_daemon_refuses_loudly_when_it_cannot_open_the_mind_and_binds_nothing() {
let root = tempfile::tempdir().unwrap();
let held = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let held = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
let path = Mind::<OwnedRedbMessagePackBackingStore>::store_path_for(root.path(), &slug(INSTANCE)).unwrap();
assert_eq!(
Daemon::open(root.path(), &slug(INSTANCE)).err(),
open_unindexed(root.path(), &slug(INSTANCE)).err(),
Some(MindRefusal::MindAlreadyOwned { path: path.display().to_string() })
);
drop(held);
let mut planted = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let mut planted = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
planted.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)])), now());
drop(planted);
let moved = Mind::<OwnedRedbMessagePackBackingStore>::store_path_for(root.path(), &slug(OTHER)).unwrap();
std::fs::create_dir_all(moved.parent().unwrap()).unwrap();
std::fs::copy(&path, &moved).unwrap();
assert_eq!(
Daemon::open(root.path(), &slug(OTHER)).err(),
open_unindexed(root.path(), &slug(OTHER)).err(),
Some(MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() })
);
}

View file

@ -183,6 +183,7 @@ mod tests {
schema_epoch: "epiphany.pipeline.epoch.v2".into(),
documents: 0,
receipts: 0,
index: huginn_mind::wire::IndexStatus::Current,
}),
"accepted",
),

View file

@ -0,0 +1,177 @@
//! In-memory ports for the projector's tests. Each fake is a handle on shared
//! state, so a test keeps observing after it has moved the fake into a
//! projector or a worker thread.
use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Condvar, Mutex};
use anyhow::{Result, bail};
use super::{CollectionMeta, Described, Embedder, ModelIdentity, Point, VectorIndex};
pub(crate) fn identity(digest: &str) -> ModelIdentity {
ModelIdentity { name: "model".into(), digest: digest.into(), dimensions: 4 }
}
/// A door a test holds shut: `embed` announces it has arrived and waits.
#[derive(Clone, Default)]
pub(crate) struct Gate(Arc<(Mutex<(u32, bool)>, Condvar)>);
impl Gate {
pub(crate) fn shut() -> Self {
Self::default()
}
/// Blocks until `count` calls to `embed` have arrived at the gate.
pub(crate) fn wait_arrived(&self, count: u32) {
let mut state = self.0.0.lock().unwrap();
while state.0 < count {
state = self.0.1.wait(state).unwrap();
}
}
pub(crate) fn open(&self) {
self.0.0.lock().unwrap().1 = true;
self.0.1.notify_all();
}
fn pass(&self) {
let mut state = self.0.0.lock().unwrap();
state.0 += 1;
self.0.1.notify_all();
while !state.1 {
state = self.0.1.wait(state).unwrap();
}
}
}
pub(crate) struct EmbedderState {
pub(crate) identity: ModelIdentity,
pub(crate) down: bool,
/// The texts of every completed call, in order.
pub(crate) embedded: Vec<Vec<String>>,
}
#[derive(Clone)]
pub(crate) struct FakeEmbedder {
pub(crate) state: Arc<Mutex<EmbedderState>>,
gate: Option<Gate>,
}
impl FakeEmbedder {
pub(crate) fn new(digest: &str) -> Self {
let state = EmbedderState { identity: identity(digest), down: false, embedded: Vec::new() };
Self { state: Arc::new(Mutex::new(state)), gate: None }
}
pub(crate) fn behind(mut self, gate: &Gate) -> Self {
self.gate = Some(gate.clone());
self
}
pub(crate) fn texts(&self) -> Vec<String> {
self.state.lock().unwrap().embedded.iter().flatten().cloned().collect()
}
}
impl Embedder for FakeEmbedder {
fn model_identity(&mut self) -> Result<ModelIdentity> {
let state = self.state.lock().unwrap();
if state.down {
bail!("the embedder is down");
}
Ok(state.identity.clone())
}
fn embed(&mut self, texts: &[String]) -> Result<Vec<Vec<f32>>> {
if let Some(gate) = &self.gate {
gate.pass();
}
let mut state = self.state.lock().unwrap();
if state.down {
bail!("the embedder is down");
}
state.embedded.push(texts.to_vec());
let dimensions = state.identity.dimensions as usize;
Ok(texts.iter().map(|text| (0..dimensions).map(|at| (text.len() + at) as f32).collect()).collect())
}
}
pub(crate) struct Stored {
pub(crate) label: Described,
pub(crate) points: BTreeMap<String, Point>,
}
#[derive(Default)]
pub(crate) struct IndexState {
pub(crate) collections: BTreeMap<String, Stored>,
pub(crate) down: bool,
pub(crate) recreated: Vec<(String, CollectionMeta)>,
/// The point ids of every upsert call, in order.
pub(crate) upserts: Vec<Vec<String>>,
}
#[derive(Clone, Default)]
pub(crate) struct FakeIndex {
pub(crate) state: Arc<Mutex<IndexState>>,
}
impl FakeIndex {
pub(crate) fn holding(&self, collection: &str) -> BTreeSet<String> {
let state = self.state.lock().unwrap();
state.collections.get(collection).map(|stored| stored.points.keys().cloned().collect()).unwrap_or_default()
}
/// Plants a collection this organ did not make, or made for another mind.
pub(crate) fn plant(&self, collection: &str, label: Described) {
self.state.lock().unwrap().collections.insert(collection.into(), Stored { label, points: BTreeMap::new() });
}
}
impl VectorIndex for FakeIndex {
fn describe(&mut self, collection: &str) -> Result<Described> {
let state = self.state.lock().unwrap();
if state.down {
bail!("the vector store is down");
}
Ok(state.collections.get(collection).map_or(Described::Absent, |stored| stored.label.clone()))
}
fn recreate(&mut self, collection: &str, meta: &CollectionMeta) -> Result<()> {
let mut state = self.state.lock().unwrap();
if state.down {
bail!("the vector store is down");
}
state.recreated.push((collection.into(), meta.clone()));
state
.collections
.insert(collection.into(), Stored { label: Described::Labelled(meta.clone()), points: BTreeMap::new() });
Ok(())
}
fn ids(&mut self, collection: &str) -> Result<BTreeSet<String>> {
let state = self.state.lock().unwrap();
if state.down {
bail!("the vector store is down");
}
Ok(state.collections.get(collection).map(|stored| stored.points.keys().cloned().collect()).unwrap_or_default())
}
fn upsert(&mut self, collection: &str, points: &[Point]) -> Result<()> {
let mut state = self.state.lock().unwrap();
if state.down {
bail!("the vector store is down");
}
state.upserts.push(points.iter().map(|point| point.id.clone()).collect());
let Some(stored) = state.collections.get_mut(collection) else { bail!("no collection {collection}") };
for point in points {
stored.points.insert(point.id.clone(), point.clone());
}
Ok(())
}
}
/// A working embedder and vector store, for tests whose subject is elsewhere.
pub(crate) fn pair() -> (FakeEmbedder, FakeIndex) {
(FakeEmbedder::new("d1"), FakeIndex::default())
}

View file

@ -0,0 +1,642 @@
//! The semantic index's write side: a projection of the mind into a vector
//! collection, kept by one worker thread.
//!
//! The mind is the truth and this is derived. `huginn-mind` decides what a
//! document's text is (`IndexEntry`); this module owns when it is embedded and
//! written, and nothing here is stored in the mind: the worker's health lives
//! in memory and a restart recomputes it.
//!
//! One owner per decision. `Projector::reconcile` is the only place a point is
//! judged missing (the mind against the collection, by set difference, at
//! startup), and the worker is the only writer of points. Live admissions
//! reach the worker through `WorkerSink::committed`, which builds the entries
//! (local and cheap), leaves them in an inbox and returns: admission never
//! waits on an embedder or a vector store. Live entries and startup entries
//! share one embed-and-upsert routine, `Projector::flush_batch`.
//!
//! The ports are traits so the projector is tested with in-memory fakes:
//! `Embedder` (`ollama`) and `VectorIndex` (`qdrant`). Those two files are the
//! xenos boundary; nothing else in the organ speaks HTTP or JSON.
//!
//! 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
//! 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.
//!
//! A collection belongs to a mind, not to a name. It is labelled with the
//! instance and with the mind's genesis receipt id (`Mind::genesis_receipt_id`),
//! and a collection labelled for another mind of the same instance name, or
//! for none, is refused and left untouched. A mind with no receipt yet has no
//! identity, so nothing is reconciled until its first admission.
//!
//! Stated limits.
//!
//! - The inbox is unbounded: it holds at most the documents admitted while the
//! index is unreachable, each of which is also in the mind.
//! - Vectors of the wrong length are refused rather than written.
//! - The real input limit is the embedding model's context, not the 16 KiB
//! `INDEX_TEXT_MAX_BYTES` bound. `qwen3-embedding:0.6b` has a 4096 token
//! context, so dense text is cut at about 4k tokens and the tail of a long
//! document is not searchable. Ollama truncates silently by default; its
//! `truncate: false` option refuses over-context input instead, which would
//! 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.
pub mod ollama;
pub mod qdrant;
#[cfg(test)]
pub(crate) mod fakes;
use std::collections::{BTreeMap, BTreeSet};
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 huginn_mind::wire::IndexStatus;
use huginn_mind::{INDEX_TEXT_VERSION, IndexEntry, Mind, MindStore};
use sha2::{Digest, Sha256};
use crate::daemon::IndexSink;
/// The `managed_by` a collection carries when this organ made it.
pub const MANAGED_BY: &str = "huginn";
/// Documents embedded and written per call.
pub const BATCH: usize = 32;
/// 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.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ModelIdentity {
pub name: String,
pub digest: String,
pub dimensions: u32,
}
/// What a collection says about how it was made.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CollectionMeta {
pub managed_by: String,
pub instance: String,
/// The mind's genesis receipt id: the identity the instance name lacks.
pub mind: String,
pub model: String,
pub model_digest: String,
pub dimensions: u32,
pub index_text_version: u32,
}
/// What the vector store holds under a collection's name.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Described {
Absent,
/// A collection that carries no metadata this organ wrote.
Unlabelled,
Labelled(CollectionMeta),
}
/// What travels beside a vector. The id of the point is derived from `doc_id`.
#[derive(Clone, Debug, PartialEq)]
pub struct PointPayload {
pub doc_id: String,
pub kind: String,
pub root: String,
pub ordinal: u64,
pub text_sha256: String,
}
#[derive(Clone, Debug, PartialEq)]
pub struct Point {
pub id: String,
pub vector: Vec<f32>,
pub payload: PointPayload,
}
pub trait Embedder {
fn model_identity(&mut self) -> Result<ModelIdentity>;
/// One vector per text, in order.
fn embed(&mut self, texts: &[String]) -> Result<Vec<Vec<f32>>>;
}
pub trait VectorIndex {
fn describe(&mut self, collection: &str) -> Result<Described>;
/// Drops the collection if it exists and creates it empty with this
/// metadata.
fn recreate(&mut self, collection: &str, meta: &CollectionMeta) -> Result<()>;
/// 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 mind's collection: one per instance.
pub fn collection_name(instance: &Slug) -> String {
format!("huginn_mind_{}", instance.0)
}
/// The point id of a document: the first sixteen bytes of the SHA-256 of the
/// document id, shaped as a version 5 UUID, which is the convention voidbot's
/// collections use. The real id travels in the payload.
pub fn point_id(document_id: &str) -> String {
let mut digest: [u8; 32] = Sha256::digest(document_id.as_bytes()).into();
digest[6] = (digest[6] & 0x0f) | 0x50;
digest[8] = (digest[8] & 0x3f) | 0x80;
let hex: String = digest[..16].iter().map(|byte| format!("{byte:02x}")).collect();
format!("{}-{}-{}-{}-{}", &hex[0..8], &hex[8..12], &hex[12..16], &hex[16..20], &hex[20..32])
}
/// Why a step did not finish. Both are retried; they differ in what the
/// operator is told.
#[derive(Debug)]
pub enum StepError {
/// The collection is another writer's and is left alone.
Refused(String),
/// The embedder or the vector store did not answer, or answered wrongly.
Unavailable(anyhow::Error),
}
impl From<anyhow::Error> for StepError {
fn from(error: anyhow::Error) -> Self {
Self::Unavailable(error)
}
}
impl std::fmt::Display for StepError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Refused(reason) => formatter.write_str(reason),
Self::Unavailable(error) => write!(formatter, "{error:#}"),
}
}
}
/// Whether a step left work to do.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Advance {
Progressed,
Idle,
}
/// The projection's state: the entries not yet known to be in the collection,
/// and, once reconciled, the length of vector it accepts.
pub struct Projector<E: Embedder, V: VectorIndex> {
embedder: E,
index: V,
instance: String,
/// `None` until the mind's first admission; nothing is reconciled before.
mind: Option<String>,
collection: String,
/// Every entry ever given, by document id: refilling a lost collection
/// needs text the mind's writes have already gone past.
known: BTreeMap<String, IndexEntry>,
/// The ids in `known` whose points the collection may lack.
wanted: BTreeSet<String>,
dimensions: Option<u32>,
}
impl<E: Embedder, V: VectorIndex> Projector<E, V> {
pub fn new(embedder: E, index: V, instance: &Slug, mind: Option<String>) -> Self {
Self {
embedder,
index,
instance: instance.0.clone(),
mind,
collection: collection_name(instance),
known: BTreeMap::new(),
wanted: BTreeSet::new(),
dimensions: None,
}
}
/// The mind's identity, once it has one. It never changes after that.
fn identify(&mut self, mind: String) {
if self.mind.is_none() {
self.mind = Some(mind);
}
}
/// Entries the collection may lack. Idempotent: an entry already wanted is
/// the same document.
pub fn want(&mut self, entries: impl IntoIterator<Item = IndexEntry>) {
for entry in entries {
self.wanted.insert(entry.id.id.0.clone());
self.known.insert(entry.id.id.0.clone(), entry);
}
}
pub fn pending(&self) -> u32 {
u32::try_from(self.wanted.len()).unwrap_or(u32::MAX)
}
/// One step: reconcile first, then one batch at a time until nothing is
/// wanted.
///
/// Any failure forgets the reconciliation: the next attempt starts from
/// the model and the collection as they are then.
pub fn advance(&mut self) -> Result<Advance, StepError> {
let stepped = self.step();
if stepped.is_err() {
self.dimensions = None;
}
stepped
}
fn step(&mut self) -> Result<Advance, StepError> {
if self.mind.is_none() {
return Ok(Advance::Idle);
}
if self.dimensions.is_none() {
self.reconcile()?;
return Ok(Advance::Progressed);
}
if self.wanted.is_empty() {
return Ok(Advance::Idle);
}
self.flush_batch()?;
Ok(Advance::Progressed)
}
/// 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),
/// refuses a collection that is not this mind's, then makes `wanted` every
/// known entry whose point the collection does not hold.
pub fn reconcile(&mut self) -> Result<(), StepError> {
let Some(mind) = self.mind.clone() else {
return Err(StepError::Unavailable(anyhow::anyhow!("the mind has no identity yet")));
};
let identity = self.embedder.model_identity().context("reading the embedding model's identity")?;
let wanted = CollectionMeta {
managed_by: MANAGED_BY.into(),
instance: self.instance.clone(),
mind: mind.clone(),
model: identity.name.clone(),
model_digest: identity.digest.clone(),
dimensions: identity.dimensions,
index_text_version: INDEX_TEXT_VERSION,
};
match self.index.describe(&self.collection).context("describing the collection")? {
Described::Absent => self.recreate(&wanted)?,
Described::Unlabelled => {
return Err(StepError::Refused(format!(
"collection {} exists and carries no Huginn metadata; it is not this organ's to rewrite",
self.collection
)));
}
Described::Labelled(found)
if found.managed_by != MANAGED_BY || found.instance != self.instance || found.mind != mind =>
{
return Err(StepError::Refused(format!(
"collection {} is managed by {:?} for instance {:?}, mind {:?}; this organ serves instance {:?}, mind {:?}",
self.collection, found.managed_by, found.instance, found.mind, self.instance, mind
)));
}
Described::Labelled(found) if found != wanted => self.recreate(&wanted)?,
Described::Labelled(_) => {}
}
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);
Ok(())
}
fn recreate(&mut self, meta: &CollectionMeta) -> Result<()> {
self.index.recreate(&self.collection, meta).context("creating the collection")
}
/// The one embed-and-upsert routine, for startup and live entries alike:
/// the first `BATCH` wanted entries are embedded, written, and only then
/// no longer wanted.
pub fn flush_batch(&mut self) -> Result<(), StepError> {
self.flush().map_err(StepError::Unavailable)
}
fn flush(&mut self) -> Result<()> {
let dimensions = self.dimensions.context("flush before reconcile")? 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() {
return Ok(());
}
let texts: Vec<String> = batch.iter().map(|entry| entry.text.clone()).collect();
let vectors = self.embedder.embed(&texts).context("embedding")?;
ensure!(vectors.len() == batch.len(), "the embedder returned {} vectors for {} texts", vectors.len(), batch.len());
let mut points = Vec::with_capacity(batch.len());
for (entry, vector) in batch.iter().zip(vectors) {
ensure!(
vector.len() == dimensions,
"the embedder returned a vector of {} for {} where the collection holds {dimensions}",
vector.len(),
entry.id.id.0
);
points.push(Point {
id: point_id(&entry.id.id.0),
vector,
payload: PointPayload {
doc_id: entry.id.id.0.clone(),
kind: entry.id.kind.name().into(),
root: entry.root.clone(),
ordinal: entry.ordinal,
text_sha256: entry.text_sha256.clone(),
},
});
}
self.index.upsert(&self.collection, &points).context("writing points")?;
for entry in &batch {
self.wanted.remove(&entry.id.id.0);
}
Ok(())
}
/// The health after a step that did not fail.
pub fn progress_status(&self) -> IndexStatus {
match (self.dimensions, self.pending()) {
(None, 0) if self.mind.is_none() => IndexStatus::Current,
(None, pending) => IndexStatus::Reconciling { pending },
(Some(_), 0) => IndexStatus::Current,
(Some(_), pending) => IndexStatus::Behind { pending },
}
}
/// The health after `attempts` consecutive failures, the last of them
/// `error`.
pub fn failure_status(&self, error: &StepError, attempts: u32) -> IndexStatus {
let pending = self.pending();
match error {
StepError::Refused(reason) => IndexStatus::Refused { pending, attempts, reason: reason.clone() },
StepError::Unavailable(error) => IndexStatus::Failing { pending, attempts, error: format!("{error:#}") },
}
}
}
/// The retry delay: it starts at `initial`, doubles per consecutive failure and
/// stops at `max`.
#[derive(Clone, Copy, Debug)]
pub struct Backoff {
pub initial: Duration,
pub max: Duration,
}
impl Default for Backoff {
fn default() -> Self {
Self { initial: Duration::from_secs(30), max: Duration::from_secs(600) }
}
}
impl Backoff {
pub fn after(&self, delay: Duration) -> Duration {
(delay * 2).min(self.max)
}
}
struct Shared {
inbox: Vec<IndexEntry>,
/// The mind's identity, once a commit has revealed it to a mind that had none.
mind: Option<String>,
status: IndexStatus,
closed: bool,
}
type Lock = Arc<(Mutex<Shared>, Condvar)>;
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>) {
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());
}
}
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;
}
}
}
}
fn publish(shared: &Lock, status: IndexStatus) {
lock(shared).status = status;
shared.1.notify_all();
}
/// The status with what the worker has not yet taken from the inbox added to
/// its pending count.
fn including(status: &IndexStatus, queued: u32) -> IndexStatus {
use IndexStatus as S;
match status {
S::Reconciling { pending } => S::Reconciling { pending: pending + queued },
S::Current if queued == 0 => S::Current,
S::Current => S::Behind { pending: queued },
S::Behind { pending } => S::Behind { pending: pending + queued },
S::Failing { pending, attempts, error } => {
S::Failing { pending: pending + queued, attempts: *attempts, error: error.clone() }
}
S::Refused { pending, attempts, reason } => {
S::Refused { pending: pending + queued, attempts: *attempts, reason: reason.clone() }
}
}
}
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;
loop {
let (mind, inbox) = {
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);
}
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() {
Ok(Advance::Progressed) => {
failures = 0;
delay = backoff.initial;
publish(&shared, projector.progress_status());
}
Ok(Advance::Idle) => {
failures = 0;
delay = backoff.initial;
publish(&shared, projector.progress_status());
wait_for_work(&shared, None);
}
Err(error) => {
failures = failures.saturating_add(1);
eprintln!(
"huginn: the index could not advance (attempt {failures}, {} pending, retrying in {:?}): {error}",
projector.pending(),
delay
);
publish(&shared, projector.failure_status(&error, failures));
wait_for_work(&shared, Some(delay));
delay = backoff.after(delay);
}
}
}
}
/// The daemon's index: a handle on the one worker thread that owns the ports.
/// `committed` builds entries and queues them; `status` reads the worker's
/// health. Dropping the sink tells the worker to stop and does not wait for
/// it: a call in flight ends with its own timeout or with the process.
pub struct WorkerSink {
shared: Lock,
/// Whether the worker has been told the mind's identity.
identified: bool,
}
impl WorkerSink {
/// Starts the worker. `mind` is the mind's genesis receipt id, `None` for
/// a mind not yet written. `startup` is every indexable document the mind
/// holds now; the worker's first act is to reconcile the collection
/// against it.
pub fn spawn<E, V>(
embedder: E,
index: V,
instance: &Slug,
mind: Option<String>,
startup: Vec<IndexEntry>,
backoff: Backoff,
) -> Self
where
E: Embedder + Send + 'static,
V: VectorIndex + Send + 'static,
{
let identified = mind.is_some();
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 }),
Condvar::new(),
));
let worker = Arc::clone(&shared);
std::thread::spawn(move || work(projector, worker, backoff));
Self { shared, identified }
}
/// Blocks until the health satisfies `predicate`, and returns it. For
/// tests and probes that must observe the worker without a clock; it gives
/// up after two minutes rather than hang a run.
pub fn wait_status(&self, predicate: impl Fn(&IndexStatus) -> bool) -> IndexStatus {
let mut guard = lock(&self.shared);
loop {
let seen = including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX));
if predicate(&seen) {
return seen;
}
let (next, result) = self
.shared
.1
.wait_timeout(guard, Duration::from_secs(120))
.unwrap_or_else(|poisoned| poisoned.into_inner());
guard = next;
assert!(!result.timed_out(), "the index never reached the awaited state; last seen {:?}", guard.status);
}
}
}
impl Drop for WorkerSink {
fn drop(&mut self) {
lock(&self.shared).closed = true;
self.shared.1.notify_all();
}
}
impl<S: MindStore> IndexSink<S> for WorkerSink {
fn committed(&mut self, mind: &Mind<S>, writes: &[PipelineRef]) -> Result<()> {
let entries = mind.index_entries(Some(writes))?;
let revealed = if self.identified { None } else { mind.genesis_receipt_id()? };
if entries.is_empty() && revealed.is_none() {
return Ok(());
}
let mut guard = lock(&self.shared);
guard.inbox.extend(entries);
if let Some(revealed) = revealed {
guard.mind = Some(revealed);
self.identified = true;
}
drop(guard);
self.shared.1.notify_all();
Ok(())
}
fn status(&self) -> IndexStatus {
let guard = lock(&self.shared);
including(&guard.status, u32::try_from(guard.inbox.len()).unwrap_or(u32::MAX))
}
}
/// Plain HTTP to one base address, in text: the adapters own what the text
/// means. There is no TLS in this build; the addresses it serves are the
/// private mesh's.
pub(crate) struct Http {
agent: ureq::Agent,
base: String,
}
#[derive(Clone, Copy)]
pub(crate) enum Method {
Get,
Put,
Post,
Delete,
}
impl Http {
pub(crate) fn new(base_url: &str, timeout: Duration) -> Self {
let agent: ureq::Agent =
ureq::Agent::config_builder().timeout_global(Some(timeout)).http_status_as_error(false).build().into();
Self { agent, base: base_url.trim_end_matches('/').to_owned() }
}
/// The status and the body, whatever the status is.
pub(crate) fn request(&self, method: Method, path: &str, body: Option<&str>) -> Result<(u16, String)> {
let url = format!("{}{path}", self.base);
let json = "application/json";
let sent = match method {
Method::Get => self.agent.get(&url).call(),
Method::Delete => self.agent.delete(&url).call(),
Method::Post => self.agent.post(&url).header("content-type", json).send(body.unwrap_or_default()),
Method::Put => self.agent.put(&url).header("content-type", json).send(body.unwrap_or_default()),
};
let mut response = sent.with_context(|| format!("no answer from {url}"))?;
let status = response.status().as_u16();
let text = response.body_mut().read_to_string().with_context(|| format!("reading the answer from {url}"))?;
Ok((status, text))
}
}
#[cfg(test)]
mod tests;

View file

@ -0,0 +1,74 @@
//! The embedder port over Ollama's HTTP API. This file and `qdrant.rs` are the
//! xenos boundary: JSON exists here and nowhere else in the organ.
use std::time::Duration;
use anyhow::{Context, Result, bail, ensure};
use serde_json::{Value, json};
use super::{Embedder, Http, Method, ModelIdentity};
/// The instruction Qwen3 embedding models take in front of a query (never in
/// front of a document). Cut 11b's semantic read prepends it as
/// `Instruct: <this>\nQuery: <text>`.
pub const QUERY_INSTRUCTION: &str =
"Given a question about a project's pipeline state, retrieve the documents that answer it";
/// One model on one Ollama. A cold model takes seconds to load, so the
/// timeout is generous; the worker owns the schedule around it.
pub struct OllamaEmbedder {
http: Http,
model: String,
}
impl OllamaEmbedder {
pub fn new(base_url: &str, model: &str) -> Self {
Self { http: Http::new(base_url, Duration::from_secs(300)), model: model.to_owned() }
}
fn ask(&self, method: Method, path: &str, body: Option<&Value>) -> Result<Value> {
let text = body.map(Value::to_string);
let (status, answer) = self.http.request(method, path, text.as_deref())?;
ensure!(status == 200, "Ollama answered {status} to {path}: {}", answer.chars().take(300).collect::<String>());
serde_json::from_str(&answer).with_context(|| format!("Ollama's answer to {path} is not JSON"))
}
}
impl Embedder for OllamaEmbedder {
/// The configured name, the digest Ollama lists for it, and the model's
/// embedding length. The address is not part of the identity.
fn model_identity(&mut self) -> Result<ModelIdentity> {
let tags = self.ask(Method::Get, "/api/tags", None)?;
let listed = tags["models"].as_array().context("Ollama's tag list has no models")?;
let digest = listed
.iter()
.find(|entry| entry["name"] == self.model.as_str() || entry["model"] == self.model.as_str())
.and_then(|entry| entry["digest"].as_str())
.with_context(|| format!("Ollama does not list the model {}", self.model))?
.to_owned();
let shown = self.ask(Method::Post, "/api/show", Some(&json!({ "model": self.model })))?;
let info = shown["model_info"].as_object().context("Ollama's model description has no model_info")?;
let Some(length) = info.iter().find(|(key, _)| key.ends_with(".embedding_length")).and_then(|(_, value)| value.as_u64())
else {
bail!("Ollama does not say how long {}'s embeddings are", self.model);
};
let dimensions = u32::try_from(length).context("an embedding length beyond u32")?;
Ok(ModelIdentity { name: self.model.clone(), digest, dimensions })
}
fn embed(&mut self, texts: &[String]) -> Result<Vec<Vec<f32>>> {
let answer = self.ask(Method::Post, "/api/embed", Some(&json!({ "model": self.model, "input": texts })))?;
let embeddings = answer["embeddings"].as_array().context("Ollama's answer has no embeddings")?;
embeddings
.iter()
.map(|vector| {
vector
.as_array()
.context("an embedding is not a list")?
.iter()
.map(|number| number.as_f64().map(|value| value as f32).context("an embedding holds a non-number"))
.collect()
})
.collect()
}
}

View file

@ -0,0 +1,135 @@
//! The vector-store port over Qdrant's HTTP API. With `ollama.rs`, the xenos
//! boundary: JSON exists here and nowhere else in the organ.
use std::collections::BTreeSet;
use std::time::Duration;
use anyhow::{Context, Result, ensure};
use serde_json::{Value, json};
use super::{CollectionMeta, Described, Http, Method, Point, VectorIndex};
/// One Qdrant. Every path names a collection this organ was told to own; it
/// never lists or touches another.
pub struct QdrantIndex {
http: Http,
}
/// The ids fetched per scroll page.
const SCROLL_PAGE: u64 = 1000;
impl QdrantIndex {
pub fn new(base_url: &str) -> Self {
Self { http: Http::new(base_url, Duration::from_secs(60)) }
}
fn ask(&self, method: Method, path: &str, body: Option<&Value>, ok: &[u16]) -> Result<(u16, Value)> {
let text = body.map(Value::to_string);
let (status, answer) = self.http.request(method, path, text.as_deref())?;
ensure!(ok.contains(&status), "Qdrant answered {status} to {path}: {}", answer.chars().take(300).collect::<String>());
let parsed = if answer.is_empty() { Value::Null } else { serde_json::from_str(&answer).unwrap_or(Value::Null) };
Ok((status, parsed))
}
}
fn metadata_json(meta: &CollectionMeta) -> Value {
json!({
"managed_by": meta.managed_by,
"instance": meta.instance,
"mind": meta.mind,
"model": meta.model,
"model_digest": meta.model_digest,
"dimensions": meta.dimensions,
"index_text_version": meta.index_text_version,
})
}
/// What a collection's `config.metadata` says. A collection whose metadata
/// does not name a writer and an instance is unlabelled; one that does but is
/// missing the rest reads as an empty mind or model, which will not match: an
/// empty mind is refused as another's, an empty model is rebuilt when the
/// writer, the instance and the mind are ours.
fn described(metadata: &Value) -> Described {
let text = |key: &str| metadata[key].as_str().map(str::to_owned);
let (Some(managed_by), Some(instance)) = (text("managed_by"), text("instance")) else {
return Described::Unlabelled;
};
let number = |key: &str| metadata[key].as_u64().and_then(|value| u32::try_from(value).ok()).unwrap_or(0);
Described::Labelled(CollectionMeta {
managed_by,
instance,
mind: text("mind").unwrap_or_default(),
model: text("model").unwrap_or_default(),
model_digest: text("model_digest").unwrap_or_default(),
dimensions: number("dimensions"),
index_text_version: number("index_text_version"),
})
}
impl VectorIndex for QdrantIndex {
fn describe(&mut self, collection: &str) -> Result<Described> {
let (status, answer) = self.ask(Method::Get, &format!("/collections/{collection}"), None, &[200, 404])?;
if status == 404 {
return Ok(Described::Absent);
}
Ok(described(&answer["result"]["config"]["metadata"]))
}
fn recreate(&mut self, collection: &str, meta: &CollectionMeta) -> Result<()> {
let path = format!("/collections/{collection}");
self.ask(Method::Delete, &format!("{path}?wait=true"), None, &[200, 404])?;
let body = json!({
"vectors": { "size": meta.dimensions, "distance": "Cosine", "on_disk": true },
"metadata": metadata_json(meta),
});
self.ask(Method::Put, &format!("{path}?wait=true"), Some(&body), &[200])?;
Ok(())
}
fn ids(&mut self, collection: &str) -> Result<BTreeSet<String>> {
let mut ids = BTreeSet::new();
let mut offset = Value::Null;
loop {
let mut body = json!({ "limit": SCROLL_PAGE, "with_payload": false, "with_vector": false });
if !offset.is_null() {
body["offset"] = offset;
}
let (_, answer) =
self.ask(Method::Post, &format!("/collections/{collection}/points/scroll"), Some(&body), &[200])?;
let points = answer["result"]["points"].as_array().context("Qdrant's scroll has no points")?;
for point in points {
ids.insert(point["id"].as_str().context("a point id is not a string")?.to_owned());
}
offset = answer["result"]["next_page_offset"].clone();
if offset.is_null() {
return Ok(ids);
}
}
}
fn upsert(&mut self, collection: &str, points: &[Point]) -> Result<()> {
let points: Vec<Value> = points
.iter()
.map(|point| {
json!({
"id": point.id,
"vector": point.vector,
"payload": {
"doc_id": point.payload.doc_id,
"kind": point.payload.kind,
"root": point.payload.root,
"ordinal": point.payload.ordinal,
"text_sha256": point.payload.text_sha256,
},
})
})
.collect();
self.ask(
Method::Put,
&format!("/collections/{collection}/points?wait=true"),
Some(&json!({ "points": points })),
&[200],
)?;
Ok(())
}
}

View file

@ -0,0 +1,753 @@
//! The projection's rules, against in-memory ports; the adapters' request and
//! answer shapes, against a stub that speaks just enough HTTP. No test needs a
//! live Qdrant or Ollama.
use std::collections::BTreeSet;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, IndexStatus};
use huginn_mind::{Mind, OwnedRedbMessagePackBackingStore, PipelineAdmissionOutcome};
use tempfile::TempDir;
use super::fakes::{FakeEmbedder, FakeIndex, Gate, pair};
use super::ollama::OllamaEmbedder;
use super::qdrant::QdrantIndex;
use super::*;
use crate::daemon::Daemon;
use crate::daemon::tests::{INSTANCE, OTHER, batch, campaign_seed, now, question_and_ruling, slug};
fn collection() -> String {
collection_name(&slug(INSTANCE))
}
/// A mind identity for tests that project entries without a mind behind them.
const MIND: &str = "mind-commit-first";
/// A mind of the same instance name whose first batch differs, so its genesis
/// receipt id does too.
fn other_mind_of_the_same_name() -> (TempDir, Mind<OwnedRedbMessagePackBackingStore>) {
let root = tempfile::tempdir().unwrap();
let mut mind = Mind::open(root.path(), &slug(INSTANCE)).unwrap();
let (question, _) = question_and_ruling();
let mut first = campaign_seed();
first.push(question);
let outcome = mind.admit(batch(INSTANCE, first), now());
assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
(root, mind)
}
/// A real mind holding a campaign, a question, the ruling that answers it and
/// the resolution admission derived: four indexable documents, beside the
/// identity and the stewardship, which carry no text.
fn mind_with_documents() -> (TempDir, Mind<OwnedRedbMessagePackBackingStore>) {
let root = tempfile::tempdir().unwrap();
let mut mind = Mind::open(root.path(), &slug(INSTANCE)).unwrap();
let (question, ruling) = question_and_ruling();
for documents in [campaign_seed(), vec![question, ruling]] {
let outcome = mind.admit(batch(INSTANCE, documents), now());
assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
}
(root, mind)
}
/// How many times the collection was created and how many upserts it took.
fn counts(index: &FakeIndex) -> (usize, usize) {
let state = index.state.lock().unwrap();
(state.recreated.len(), state.upserts.len())
}
fn drain<E: Embedder, V: VectorIndex>(projector: &mut Projector<E, V>) {
while projector.advance().unwrap() == Advance::Progressed {}
}
fn point_ids(entries: &[IndexEntry]) -> BTreeSet<String> {
entries.iter().map(|entry| point_id(&entry.id.id.0)).collect()
}
fn projected(embedder: &FakeEmbedder, index: &FakeIndex, entries: &[IndexEntry]) -> Projector<FakeEmbedder, FakeIndex> {
let mut projector = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
projector.want(entries.iter().cloned());
drain(&mut projector);
projector
}
/// P4's function, as voidbot's `toQdrantPointId` computes it: the first sixteen
/// bytes of the SHA-256, version and variant bits set. The fixed vectors were
/// computed independently of this crate.
#[test]
fn point_ids_are_a_pure_function_of_the_document_id() {
assert_eq!(point_id("eureka-state:question:Q1"), "1e8a661b-4b7e-5245-a9ad-b15a3fa27031");
assert_eq!(point_id("eureka-state:campaign:self"), "d4135998-051d-57fb-9360-82964874fbc9");
assert_eq!(point_id("eureka-state:question:Q1"), point_id("eureka-state:question:Q1"));
assert_ne!(point_id("eureka-state:question:Q1"), point_id("eureka-state:question:Q2"));
}
/// The projection converges by set difference: after a crash that lost two
/// points, a restart embeds and writes those two, and nothing else, and the
/// collection is not rebuilt.
#[test]
fn reconcile_indexes_exactly_the_indexable_documents_the_collection_lacks() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
assert_eq!(entries.len(), 4);
let (embedder, index) = pair();
projected(&embedder, &index, &entries);
assert_eq!(
index.holding(&collection()),
point_ids(&entries),
"one point per indexable document, none for the identity or the stewardship"
);
let lost: Vec<IndexEntry> = entries.iter().take(2).cloned().collect();
{
let mut state = index.state.lock().unwrap();
let stored = state.collections.get_mut(&collection()).unwrap();
for entry in &lost {
stored.points.remove(&point_id(&entry.id.id.0));
}
state.upserts.clear();
state.recreated.clear();
}
let restarted = FakeEmbedder::new("d1");
projected(&restarted, &index, &entries);
let state = index.state.lock().unwrap();
assert!(state.recreated.is_empty(), "a collection that agrees is not rebuilt");
let written: BTreeSet<String> = state.upserts.concat().into_iter().collect();
assert_eq!(written, point_ids(&lost), "exactly the missing points were written");
assert_eq!(state.upserts.concat().len(), 2, "each once");
assert_eq!(restarted.texts().len(), 2, "and only their text was embedded");
drop(state);
assert_eq!(index.holding(&collection()), point_ids(&entries));
}
/// Compatibility is the model by name and digest, its dimensions and the text
/// version. Any of them differing rebuilds the collection whole, and no vector
/// of one model survives beside another's.
#[test]
fn a_model_digest_change_rebuilds_the_collection() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let changes: [(&str, Box<dyn Fn(&mut ModelIdentity)>); 3] = [
("digest", Box::new(|identity| identity.digest = "d2".into())),
("name", Box::new(|identity| identity.name = "other-model".into())),
("dimensions", Box::new(|identity| identity.dimensions = 8)),
];
for (what, change) in changes {
let (embedder, index) = pair();
projected(&embedder, &index, &entries);
assert_eq!(index.state.lock().unwrap().recreated.len(), 1, "the first run creates the collection");
let second = FakeEmbedder::new("d1");
change(&mut second.state.lock().unwrap().identity);
let expected = second.state.lock().unwrap().identity.clone();
projected(&second, &index, &entries);
let state = index.state.lock().unwrap();
assert_eq!(state.recreated.len(), 2, "a changed {what} rebuilds");
let (_, meta) = state.recreated.last().unwrap();
assert_eq!(
(&meta.model, &meta.model_digest, meta.dimensions),
(&expected.name, &expected.digest, expected.dimensions),
"{what}"
);
assert_eq!(second.texts().len(), 4, "every document is embedded again under {what}");
assert_eq!(state.collections[&collection()].points.len(), 4);
}
// The same model, digest and dimensions: nothing is rebuilt or rewritten.
let (embedder, index) = pair();
projected(&embedder, &index, &entries);
let same = FakeEmbedder::new("d1");
projected(&same, &index, &entries);
assert_eq!(index.state.lock().unwrap().recreated.len(), 1);
assert!(same.texts().is_empty());
// A text version the collection was not built with is a rebuild too.
{
let mut state = index.state.lock().unwrap();
let Described::Labelled(meta) = &mut state.collections.get_mut(&collection()).unwrap().label else { panic!() };
meta.index_text_version += 1;
}
projected(&FakeEmbedder::new("d1"), &index, &entries);
assert_eq!(index.state.lock().unwrap().recreated.len(), 2);
}
/// No cross-mind writes: a collection that another writer or another instance
/// made, or that carries no label, is left exactly as it was.
#[test]
fn a_collection_owned_by_another_instance_is_never_recreated() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let ours = |managed_by: &str, instance: &str, mind: &str| CollectionMeta {
managed_by: managed_by.into(),
instance: instance.into(),
mind: mind.into(),
model: "model".into(),
model_digest: "d1".into(),
dimensions: 4,
index_text_version: INDEX_TEXT_VERSION,
};
for foreign in [
Described::Unlabelled,
Described::Labelled(ours(MANAGED_BY, OTHER, MIND)),
Described::Labelled(ours("voidbot", INSTANCE, MIND)),
Described::Labelled(ours(MANAGED_BY, INSTANCE, "mind-commit-another")),
Described::Labelled(ours(MANAGED_BY, INSTANCE, "")),
] {
let (embedder, index) = pair();
index.plant(&collection(), foreign.clone());
let sentinel = point_id("someone-elses-document");
index.state.lock().unwrap().collections.get_mut(&collection()).unwrap().points.insert(
sentinel.clone(),
Point {
id: sentinel.clone(),
vector: vec![0.0; 4],
payload: PointPayload {
doc_id: "someone-elses-document".into(),
kind: "x".into(),
root: "x".into(),
ordinal: 1,
text_sha256: "x".into(),
},
},
);
let mut projector = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
projector.want(entries.iter().cloned());
let error = projector.reconcile().expect_err("a foreign collection is refused");
assert!(matches!(error, StepError::Refused(_)), "{foreign:?}: {error}");
assert!(matches!(projector.failure_status(&error, 1), IndexStatus::Refused { pending: 4, .. }));
let state = index.state.lock().unwrap();
assert!(state.recreated.is_empty() && state.upserts.is_empty(), "{foreign:?} was written to");
assert_eq!(state.collections[&collection()].label, foreign);
assert!(state.collections[&collection()].points.contains_key(&sentinel), "{foreign:?} lost a point");
assert!(embedder.texts().is_empty());
}
}
/// A vector of the wrong length is refused before anything is written, and the
/// documents stay wanted.
#[test]
fn a_vector_of_the_wrong_length_is_refused_and_nothing_is_dropped() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
let mut projector = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
projector.want(entries.iter().cloned());
projector.reconcile().unwrap();
embedder.state.lock().unwrap().identity.dimensions = 8;
let error = projector.flush_batch().expect_err("a longer vector is refused");
assert!(matches!(error, StepError::Unavailable(_)), "{error}");
assert_eq!(projector.pending(), 4);
assert!(index.state.lock().unwrap().upserts.is_empty());
}
fn whoami_index(daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, WorkerSink>) -> IndexStatus {
match daemon.handle(HuginnMindRequest::Whoami, now()) {
HuginnMindResponse::Whoami(status) => status.index,
other => panic!("expected a status, got {other:?}"),
}
}
fn committed(response: HuginnMindResponse) {
assert!(
matches!(response, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })),
"{response:?}"
);
}
fn quick() -> Backoff {
Backoff { initial: Duration::from_millis(1), max: Duration::from_millis(2) }
}
/// The index never delays or refuses admission: the embedder is shut, the
/// worker is inside it, and two more admissions are answered at once. The gate
/// counts arrivals, so this observes where the work is and not how long it
/// took.
#[test]
fn an_admission_replies_while_the_embedder_is_blocked() {
let root = tempfile::tempdir().unwrap();
let mind = Mind::open(root.path(), &slug(INSTANCE)).unwrap();
let gate = Gate::shut();
let embedder = FakeEmbedder::new("d1").behind(&gate);
let index = FakeIndex::default();
let sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), mind.genesis_receipt_id().unwrap(), mind.index_entries(None).unwrap(), quick());
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()));
gate.wait_arrived(1);
let (question, ruling) = question_and_ruling();
committed(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![question, ruling])), now()));
assert_eq!(
whoami_index(&mut daemon),
IndexStatus::Behind { pending: 4 },
"the campaign is in the embedder and three more are queued"
);
assert!(index.holding(&collection()).is_empty(), "nothing has been written yet");
gate.open();
daemon.index().wait_status(|status| *status == IndexStatus::Current);
assert_eq!(index.holding(&collection()).len(), 4);
assert_eq!(whoami_index(&mut daemon), IndexStatus::Current);
}
/// A failing embedder is reported by `whoami`, retried on the backoff, and
/// nothing is dropped: when it returns, every document is written.
#[test]
fn a_failing_embedder_is_visible_in_whoami_and_retried_never_dropped() {
let root = tempfile::tempdir().unwrap();
let mind = Mind::open(root.path(), &slug(INSTANCE)).unwrap();
let (embedder, index) = pair();
let sink = WorkerSink::spawn(embedder.clone(), index.clone(), &slug(INSTANCE), None, Vec::new(), quick());
let mut daemon = Daemon::new(mind, sink);
daemon.index().wait_status(|status| *status == IndexStatus::Current);
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()));
}
let seen = daemon.index().wait_status(|status| matches!(status, IndexStatus::Failing { attempts, .. } if *attempts >= 2));
let IndexStatus::Failing { pending, error, .. } = seen else { panic!() };
assert_eq!(pending, 4);
assert!(error.contains("the embedder is down"), "{error}");
assert!(matches!(whoami_index(&mut daemon), IndexStatus::Failing { .. }));
assert!(index.holding(&collection()).is_empty());
embedder.state.lock().unwrap().down = false;
daemon.index().wait_status(|status| *status == IndexStatus::Current);
assert_eq!(index.holding(&collection()).len(), 4, "every admitted document was kept and written");
}
/// A collection that is not this mind's is reported as refused, never
/// recreated, and re-read on the backoff: once it is gone the projection
/// completes.
#[test]
fn a_refused_collection_is_reported_and_reread() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let (embedder, index) = pair();
index.plant(&collection(), Described::Unlabelled);
let sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), mind.genesis_receipt_id().unwrap(), entries, quick());
let mut daemon = Daemon::new(mind, sink);
daemon.index().wait_status(|status| matches!(status, IndexStatus::Refused { attempts, .. } if *attempts >= 2));
assert!(matches!(whoami_index(&mut daemon), IndexStatus::Refused { pending: 4, .. }));
assert!(index.state.lock().unwrap().recreated.is_empty());
index.state.lock().unwrap().collections.remove(&collection());
daemon.index().wait_status(|status| *status == IndexStatus::Current);
assert_eq!(index.holding(&collection()).len(), 4);
}
// The adapters, against a stub that answers the routes it is given.
struct Stub {
url: String,
seen: Arc<Mutex<Vec<(String, String, String)>>>,
}
type Route = (&'static str, u16, &'static str);
fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
haystack.windows(needle.len()).position(|window| window == needle)
}
/// Serves each route once, in order, then repeats the last route that matched.
/// A route matches `"<METHOD> <path>"` by prefix.
fn stub(routes: &[Route]) -> Stub {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let seen = Arc::new(Mutex::new(Vec::new()));
let routes: Arc<Mutex<Vec<(Route, bool)>>> = Arc::new(Mutex::new(routes.iter().map(|route| (*route, false)).collect()));
let log = Arc::clone(&seen);
std::thread::spawn(move || {
for stream in listener.incoming() {
let Ok(stream) = stream else { break };
let (routes, log) = (Arc::clone(&routes), Arc::clone(&log));
std::thread::spawn(move || answer_connection(stream, &routes, &log));
}
});
Stub { url, seen }
}
fn answer_connection(
mut stream: TcpStream,
routes: &Mutex<Vec<(Route, bool)>>,
log: &Mutex<Vec<(String, String, String)>>,
) {
let mut buffer: Vec<u8> = Vec::new();
loop {
let header_end = loop {
if let Some(at) = find(&buffer, b"\r\n\r\n") {
break at + 4;
}
let mut chunk = [0_u8; 4096];
match stream.read(&mut chunk) {
Ok(0) | Err(_) => return,
Ok(read) => buffer.extend_from_slice(&chunk[..read]),
}
};
let head = String::from_utf8_lossy(&buffer[..header_end]).into_owned();
let length = head
.lines()
.find_map(|line| line.to_ascii_lowercase().strip_prefix("content-length:").map(|value| value.trim().parse::<usize>().unwrap()))
.unwrap_or(0);
while buffer.len() < header_end + length {
let mut chunk = [0_u8; 4096];
match stream.read(&mut chunk) {
Ok(0) | Err(_) => return,
Ok(read) => buffer.extend_from_slice(&chunk[..read]),
}
}
let body = String::from_utf8_lossy(&buffer[header_end..header_end + length]).into_owned();
buffer.drain(..header_end + length);
let mut first = head.lines().next().unwrap().split(' ');
let (method, path) = (first.next().unwrap().to_owned(), first.next().unwrap().to_owned());
log.lock().unwrap().push((method.clone(), path.clone(), body));
let key = format!("{method} {path}");
let (status, answer) = {
let mut routes = routes.lock().unwrap();
let matching: Vec<usize> =
routes.iter().enumerate().filter(|(_, ((prefix, ..), _))| key.starts_with(prefix)).map(|(at, _)| at).collect();
let chosen = matching.iter().copied().find(|at| !routes[*at].1).or(matching.last().copied());
match chosen {
Some(at) => {
routes[at].1 = true;
(routes[at].0.1, routes[at].0.2)
}
None => (404, "{}"),
}
};
let response =
format!("HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\r\n{answer}", answer.len());
if stream.write_all(response.as_bytes()).is_err() {
return;
}
}
}
const TAGS: &str = r#"{"models":[{"name":"other:1","model":"other:1","digest":"zzz"},{"name":"m:1","model":"m:1","digest":"ac6d"}]}"#;
const SHOW: &str = r#"{"model_info":{"general.architecture":"qwen3","qwen3.embedding_length":4}}"#;
const EMBED: &str = r#"{"embeddings":[[0.5,1.0,2.0,3.0],[0.5,1.0,2.0,3.0],[0.5,1.0,2.0,3.0],[0.5,1.0,2.0,3.0]]}"#;
#[test]
fn ollama_identity_is_the_listed_digest_and_the_models_embedding_length() {
let ollama = stub(&[("GET /api/tags", 200, TAGS), ("POST /api/show", 200, SHOW), ("POST /api/embed", 200, EMBED)]);
let mut embedder = OllamaEmbedder::new(&ollama.url, "m:1");
assert_eq!(embedder.model_identity().unwrap(), ModelIdentity { name: "m:1".into(), digest: "ac6d".into(), dimensions: 4 });
let vectors = embedder.embed(&["a".into(), "b".into(), "c".into(), "d".into()]).unwrap();
assert_eq!(vectors.len(), 4);
assert_eq!(vectors[0], vec![0.5, 1.0, 2.0, 3.0]);
let seen = ollama.seen.lock().unwrap();
let (_, _, body) = seen.iter().find(|(_, path, _)| path == "/api/embed").unwrap();
assert!(body.contains("\"model\":\"m:1\"") && body.contains("\"input\":[\"a\",\"b\",\"c\",\"d\"]"), "{body}");
drop(seen);
// A listing entry that carries only `name` is still the model.
let bare = stub(&[
("GET /api/tags", 200, r#"{"models":[{"name":"m:1","digest":"ac6d"}]}"#),
("POST /api/show", 200, SHOW),
]);
assert_eq!(OllamaEmbedder::new(&bare.url, "m:1").model_identity().unwrap().digest, "ac6d");
let mut absent = OllamaEmbedder::new(&ollama.url, "not-pulled:1");
assert!(format!("{:#}", absent.model_identity().unwrap_err()).contains("does not list the model"));
let refusing = stub(&[("POST /api/embed", 500, r#"{"error":"boom"}"#)]);
assert!(OllamaEmbedder::new(&refusing.url, "m:1").embed(&["a".into()]).is_err());
}
/// Compatibility never reads the address: the same model at another address
/// leaves the collection alone, and the same address serving another digest
/// rebuilds it.
#[test]
fn the_embedders_address_is_not_part_of_the_collections_compatibility() {
let (_root, mind) = mind_with_documents();
let entries = mind.index_entries(None).unwrap();
let routes = [("GET /api/tags", 200, TAGS), ("POST /api/show", 200, SHOW), ("POST /api/embed", 200, EMBED)];
let repulled = TAGS.replace("ac6d", "beef");
let index = FakeIndex::default();
let mut first = Projector::new(OllamaEmbedder::new(&stub(&routes).url, "m:1"), index.clone(), &slug(INSTANCE), Some(MIND.into()));
first.want(entries.iter().cloned());
drain(&mut first);
assert_eq!(counts(&index), (1, 1));
// Another address, the same model and digest.
let mut moved = Projector::new(OllamaEmbedder::new(&stub(&routes).url, "m:1"), index.clone(), &slug(INSTANCE), Some(MIND.into()));
moved.want(entries.iter().cloned());
drain(&mut moved);
assert_eq!(counts(&index), (1, 1));
// The same model name re-pulled: another digest.
let leaked: &'static str = Box::leak(repulled.into_boxed_str());
let other = [("GET /api/tags", 200, leaked), routes[1], routes[2]];
let mut rebuilt = Projector::new(OllamaEmbedder::new(&stub(&other).url, "m:1"), index.clone(), &slug(INSTANCE), Some(MIND.into()));
rebuilt.want(entries.iter().cloned());
drain(&mut rebuilt);
assert_eq!(counts(&index), (2, 2));
}
#[test]
fn qdrant_describes_creates_lists_and_writes_what_the_projector_asks() {
let name = collection();
let absent = stub(&[("GET /collections/", 404, r#"{"status":{"error":"not found"}}"#)]);
assert_eq!(QdrantIndex::new(&absent.url).describe(&name).unwrap(), Described::Absent);
let unlabelled = stub(&[("GET /collections/", 200, r#"{"result":{"config":{"metadata":null}}}"#)]);
assert_eq!(QdrantIndex::new(&unlabelled.url).describe(&name).unwrap(), Described::Unlabelled);
let voidbot = stub(&[("GET /collections/", 200, r#"{"result":{"config":{"metadata":{"managedBy":"voidbot"}}}}"#)]);
assert_eq!(QdrantIndex::new(&voidbot.url).describe(&name).unwrap(), Described::Unlabelled);
let labelled = stub(&[(
"GET /collections/",
200,
r#"{"result":{"config":{"metadata":{"managed_by":"huginn","instance":"yggdrasil","mind":"mind-commit-first","model":"m:1","model_digest":"ac6d","dimensions":4,"index_text_version":1}}}}"#,
)]);
assert_eq!(
QdrantIndex::new(&labelled.url).describe(&name).unwrap(),
Described::Labelled(CollectionMeta {
managed_by: "huginn".into(),
instance: "yggdrasil".into(),
mind: "mind-commit-first".into(),
model: "m:1".into(),
model_digest: "ac6d".into(),
dimensions: 4,
index_text_version: 1,
})
);
// A scroll is followed to its last page.
let paged = stub(&[
("POST /collections/", 200, r#"{"result":{"points":[{"id":"a"},{"id":"b"}],"next_page_offset":"b"}}"#),
("POST /collections/", 200, r#"{"result":{"points":[{"id":"c"}],"next_page_offset":null}}"#),
]);
let ids = QdrantIndex::new(&paged.url).ids(&name).unwrap();
let expected: BTreeSet<String> = ["a", "b", "c"].map(String::from).into_iter().collect();
assert_eq!(ids, expected);
let seen = paged.seen.lock().unwrap();
assert!(!seen[0].2.contains("offset") && seen[1].2.contains("\"offset\":\"b\""), "{seen:?}");
drop(seen);
// A rebuild drops, then creates with the metadata and the vector shape.
let writing = stub(&[("DELETE /collections/", 200, "{}"), ("PUT /collections/", 200, "{}")]);
let meta = CollectionMeta {
managed_by: MANAGED_BY.into(),
instance: INSTANCE.into(),
mind: MIND.into(),
model: "m:1".into(),
model_digest: "ac6d".into(),
dimensions: 4,
index_text_version: INDEX_TEXT_VERSION,
};
let mut store = QdrantIndex::new(&writing.url);
store.recreate(&name, &meta).unwrap();
let point = Point {
id: point_id("d"),
vector: vec![1.0, 2.0, 3.0, 4.0],
payload: PointPayload { doc_id: "d".into(), kind: "question".into(), root: "eureka-state".into(), ordinal: 7, text_sha256: "h".into() },
};
store.upsert(&name, &[point]).unwrap();
let seen = writing.seen.lock().unwrap();
let paths: Vec<String> = seen.iter().map(|(method, path, _)| format!("{method} {path}")).collect();
assert_eq!(
paths,
[
format!("DELETE /collections/{name}?wait=true"),
format!("PUT /collections/{name}?wait=true"),
format!("PUT /collections/{name}/points?wait=true"),
]
);
for expected in ["\"size\":4", "\"distance\":\"Cosine\"", "\"on_disk\":true", "\"managed_by\":\"huginn\"", "\"index_text_version\":1"] {
assert!(seen[1].2.contains(expected), "{expected} in {}", seen[1].2);
}
for expected in ["\"doc_id\":\"d\"", "\"kind\":\"question\"", "\"root\":\"eureka-state\"", "\"ordinal\":7", "\"text_sha256\":\"h\""] {
assert!(seen[2].2.contains(expected), "{expected} in {}", seen[2].2);
}
let failing = stub(&[("PUT /collections/", 500, r#"{"status":{"error":"disk full"}}"#)]);
assert!(QdrantIndex::new(&failing.url).upsert(&name, &[]).is_err());
}
/// The collection's name is the contract with the store: one per instance,
/// `huginn_mind_<instance>`.
#[test]
fn the_collection_is_named_for_the_instance() {
assert_eq!(collection_name(&slug("yggdrasil")), "huginn_mind_yggdrasil");
assert_eq!(collection_name(&slug("a.b")), "huginn_mind_a.b");
}
#[test]
fn a_step_error_says_what_stopped_it() {
assert_eq!(StepError::Refused("not ours".into()).to_string(), "not ours");
let unavailable = StepError::Unavailable(anyhow::anyhow!("inner").context("outer"));
assert_eq!(unavailable.to_string(), "outer: inner");
}
/// What the sink reports is the worker's health with the inbox added to the
/// pending count, and a current index with something queued is behind.
#[test]
fn queued_entries_count_toward_every_status_that_has_a_pending_count() {
use IndexStatus as S;
assert_eq!(including(&S::Current, 0), S::Current);
assert_eq!(including(&S::Current, 3), S::Behind { pending: 3 });
assert_eq!(including(&S::Reconciling { pending: 2 }, 3), S::Reconciling { pending: 5 });
assert_eq!(including(&S::Behind { pending: 2 }, 3), S::Behind { pending: 5 });
assert_eq!(
including(&S::Failing { pending: 2, attempts: 4, error: "e".into() }, 3),
S::Failing { pending: 5, attempts: 4, error: "e".into() }
);
assert_eq!(
including(&S::Refused { pending: 2, attempts: 4, reason: "r".into() }, 3),
S::Refused { pending: 5, attempts: 4, reason: "r".into() }
);
}
/// Dropping the sink stops the worker: the worker's hold on the shared state
/// is released. A worker that never stopped would hold it forever.
#[test]
fn dropping_the_sink_stops_the_worker() {
let (embedder, index) = pair();
let sink = WorkerSink::spawn(embedder, index, &slug(INSTANCE), Some(MIND.into()), Vec::new(), quick());
sink.wait_status(|status| *status == IndexStatus::Current);
let shared = Arc::downgrade(&sink.shared);
drop(sink);
while shared.upgrade().is_some() {
std::thread::yield_now();
}
}
/// F1. The promised curve, pure: 30 s doubling to a 10 min cap, never below
/// the first delay and never past the cap. A retry that shrank would hot-loop
/// against a dead Ollama.
#[test]
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)));
let mut delay = backoff.initial;
let mut seen = vec![delay.as_secs()];
for _ in 0..7 {
let next = backoff.after(delay);
assert!(next >= delay && next >= backoff.initial, "the delay never shrinks: {delay:?} -> {next:?}");
assert!(next <= backoff.max, "the delay never passes the cap: {next:?}");
delay = next;
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) };
assert_eq!(small.after(Duration::from_millis(1)), Duration::from_millis(2));
assert_eq!(small.after(Duration::from_millis(2)), Duration::from_millis(3));
}
/// F2. A collection lost while the daemon runs is found by the next failed
/// write, recreated and refilled from what the projector knows, without a
/// restart.
#[test]
fn a_collection_lost_after_startup_is_recreated_and_refilled() {
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]);
assert_eq!(projector.progress_status(), IndexStatus::Current);
assert_eq!(index.holding(&collection()).len(), 3);
index.state.lock().unwrap().collections.remove(&collection());
projector.want(entries[3..].iter().cloned());
let error = projector.advance().expect_err("the write meets no collection");
assert!(matches!(error, StepError::Unavailable(_)), "{error}");
assert!(matches!(projector.failure_status(&error, 1), IndexStatus::Failing { pending: 1, .. }));
drain(&mut projector);
assert_eq!(index.state.lock().unwrap().recreated.len(), 2, "the collection was made again");
assert_eq!(index.holding(&collection()), point_ids(&entries), "and holds every document, not only the new one");
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.
#[test]
fn a_model_that_changes_dimensions_after_startup_rebuilds_the_collection() {
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.dimensions = 8;
projector.want(entries[3..].iter().cloned());
assert!(projector.advance().is_err(), "a vector of 8 does not fit a collection of 4");
drain(&mut projector);
let state = index.state.lock().unwrap();
assert_eq!(state.recreated.len(), 2);
assert_eq!(state.recreated.last().unwrap().1.dimensions, 8);
let stored = &state.collections[&collection()];
assert_eq!(stored.points.len(), 4);
assert!(stored.points.values().all(|point| point.vector.len() == 8));
}
/// F3. Two minds that share an instance name never share a collection: the
/// second is refused, and the first's points, label and history are untouched.
#[test]
fn a_second_mind_of_the_same_name_is_refused_and_leaves_the_first_alone() {
let (_first_root, first) = mind_with_documents();
let (_second_root, second) = other_mind_of_the_same_name();
let (first_id, second_id) = (first.genesis_receipt_id().unwrap().unwrap(), second.genesis_receipt_id().unwrap().unwrap());
assert_ne!(first_id, second_id);
let (embedder, index) = pair();
let mut ours = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(first_id.clone()));
ours.want(first.index_entries(None).unwrap());
drain(&mut ours);
let held = index.holding(&collection());
assert_eq!(held.len(), 4);
let intruder = FakeEmbedder::new("d1");
let mut theirs = Projector::new(intruder.clone(), index.clone(), &slug(INSTANCE), Some(second_id));
theirs.want(second.index_entries(None).unwrap());
let error = theirs.advance().expect_err("another mind's collection is refused");
assert!(matches!(error, StepError::Refused(_)), "{error}");
assert!(matches!(theirs.failure_status(&error, 1), IndexStatus::Refused { .. }));
let state = index.state.lock().unwrap();
assert_eq!(state.recreated.len(), 1, "never rebuilt by the second mind");
assert_eq!(state.upserts.len(), 1, "never written by the second mind");
assert!(intruder.texts().is_empty());
let Described::Labelled(label) = &state.collections[&collection()].label else { panic!() };
assert_eq!(label.mind, first_id);
drop(state);
assert_eq!(index.holding(&collection()), held);
}
/// A mind with no receipt has no identity: nothing is reconciled and no
/// collection is made, until the first admission reveals one.
#[test]
fn a_mind_without_an_identity_makes_no_collection_until_its_first_admission() {
let root = tempfile::tempdir().unwrap();
let mind = Mind::open(root.path(), &slug(INSTANCE)).unwrap();
assert_eq!(mind.genesis_receipt_id().unwrap(), None);
let (embedder, index) = pair();
let mut projector = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), None);
assert_eq!(projector.progress_status(), IndexStatus::Current);
assert_eq!(projector.advance().unwrap(), Advance::Idle);
let identified = Projector::new(embedder.clone(), index.clone(), &slug(INSTANCE), Some(MIND.into()));
assert_eq!(identified.progress_status(), IndexStatus::Reconciling { pending: 0 }, "an identified mind has yet to reconcile");
assert!(index.state.lock().unwrap().collections.is_empty());
let sink = WorkerSink::spawn(embedder, index.clone(), &slug(INSTANCE), None, Vec::new(), quick());
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()));
for _ in 0..500 {
if index.state.lock().unwrap().collections.contains_key(&collection()) {
break;
}
std::thread::sleep(Duration::from_millis(10));
}
let state = index.state.lock().unwrap();
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());
}

View file

@ -1,8 +1,9 @@
//! `huginn-daemon`: one process that serves one mind over CultNet.
//!
//! The crate owns four things and no rule. `Daemon::open` owns "may this
//! process serve this mind", by delegating to `Mind::open` and refusing before
//! anything listens. `Daemon::handle` owns "which `Mind` method answers this
//! The crate owns five things and no rule. `startup` owns "may this process
//! serve this mind", by delegating to `Mind::open` and refusing before
//! anything listens. `index` owns when the mind's documents are embedded and
//! written to the vector store, off the serving path. `Daemon::handle` owns "which `Mind` method answers this
//! request": it dispatches, it asks `Mind::require_instance` for every read
//! that names an instance, and it compares nothing itself. `envelope` owns "is
//! this a request at all": a message it cannot decode is answered with a typed
@ -17,8 +18,9 @@
pub mod bodies;
pub mod daemon;
pub mod envelope;
pub mod index;
pub mod serve;
pub use daemon::{Daemon, IndexSink, NoIndex};
pub use daemon::{Daemon, IndexSink, runtime_id};
pub use envelope::{FAILURE_SCHEMA, OperationFailure};
pub use serve::{Options, ServeOptions, parse_options, run, startup};

View file

@ -6,11 +6,16 @@ use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use anyhow::{Context, Result};
use huginn_daemon::index::Backoff;
use huginn_daemon::index::ollama::OllamaEmbedder;
use huginn_daemon::index::qdrant::QdrantIndex;
use huginn_daemon::serve::{ServeOptions, parse_options, run, startup};
fn main() -> Result<()> {
let options = parse_options(std::env::args().skip(1))?;
let (mut daemon, mut hub, registry) = match startup(&options) {
let embedder = OllamaEmbedder::new(&options.ollama_url, &options.embedding_model);
let index = QdrantIndex::new(&options.qdrant_url);
let (mut daemon, mut hub, registry) = match startup(&options, embedder, index, Backoff::default()) {
Ok(started) => started,
Err(error) => {
eprintln!("huginn-daemon refuses to serve {}: {error:#}", options.instance.0);

View file

@ -73,13 +73,14 @@ use huginn_mind::wire::{
DeferredAnswer, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA,
MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID,
};
use huginn_mind::{MindRefusal, MindStore, OwnedRedbMessagePackBackingStore};
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, NoIndex};
use crate::daemon::{Daemon, IndexSink, runtime_id};
use crate::envelope::{OperationFailure, decode_request, encode_failure, encode_response};
use crate::index::{Backoff, Embedder, VectorIndex, WorkerSink};
/// What the loop does when nothing is waiting. A deferred body untouched for
/// `deferred_ttl`, twice the session timeout, is dropped.
@ -99,22 +100,32 @@ impl Default for ServeOptions {
}
}
/// The three things the operator supplies. Nothing here reads the environment:
/// how Idunn supplies `--bind` is Cut 14's.
/// The six things the operator supplies. Nothing here reads the environment:
/// how Idunn supplies `--bind` is Cut 14's. The index has no off switch: an
/// unreachable Qdrant or Ollama is a degraded projection, reported by
/// `whoami`, not a configuration.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Options {
pub state_root: PathBuf,
pub instance: Slug,
pub bind: SocketAddr,
pub qdrant_url: String,
pub ollama_url: String,
pub embedding_model: String,
}
/// Exactly `--state-root`, `--instance` and `--bind`, each required, each once.
/// Exactly `--state-root`, `--instance`, `--bind`, `--qdrant-url`,
/// `--ollama-url` and `--embedding-model`, each required, each once. The two
/// URLs are plain `http://`: this build carries no TLS.
pub fn parse_options(args: impl Iterator<Item = String>) -> Result<Options> {
let mut args = args;
let mut values: BTreeMap<String, String> = BTreeMap::new();
while let Some(name) = args.next() {
let name = name.strip_prefix("--").with_context(|| format!("expected --option, got {name:?}"))?;
ensure!(matches!(name, "state-root" | "instance" | "bind"), "unsupported Huginn option --{name}");
ensure!(
matches!(name, "state-root" | "instance" | "bind" | "qdrant-url" | "ollama-url" | "embedding-model"),
"unsupported Huginn option --{name}"
);
let value = args.next().with_context(|| format!("missing value for --{name}"))?;
ensure!(values.insert(name.to_owned(), value).is_none(), "duplicate Huginn option --{name}");
}
@ -124,22 +135,44 @@ pub fn parse_options(args: impl Iterator<Item = String>) -> Result<Options> {
let state_root = PathBuf::from(take("state-root")?);
ensure!(state_root.is_absolute(), "--state-root must be an absolute path");
let bind = take("bind")?;
let url = |name: &str| -> Result<String> {
let value = take(name)?;
ensure!(value.starts_with("http://"), "--{name} must be a plain http:// address, got {value:?}");
Ok(value)
};
let embedding_model = take("embedding-model")?;
ensure!(!embedding_model.is_empty(), "--embedding-model must name a model");
Ok(Options {
state_root,
instance: Slug(take("instance")?),
bind: bind.parse().with_context(|| format!("--bind must be an ip:port, got {bind:?}"))?,
qdrant_url: url("qdrant-url")?,
ollama_url: url("ollama-url")?,
embedding_model,
})
}
/// Ruling 15, as an order: the mind opens first, and only a mind that opened
/// gets a socket. This is the one place both happen, so nothing can bind ahead
/// of the refusal.
pub fn startup(
/// gets a socket, and only a daemon that is going to serve gets an index
/// worker. This is the one place all three happen, so nothing can bind ahead
/// of the refusal and nothing reaches the vector store for a process that
/// never listens. The worker's first act is to reconcile the collection
/// against every indexable document the mind holds now.
pub fn startup<E, V>(
options: &Options,
) -> Result<(Daemon<OwnedRedbMessagePackBackingStore, NoIndex>, CultNetRudpServerHub, CultNetSchemaRegistry)> {
let daemon = Daemon::open(&options.state_root, &options.instance)?;
let hub = bind(options.bind, &daemon.runtime_id())?;
Ok((daemon, hub, schema_registry()?))
embedder: E,
index: V,
backoff: Backoff,
) -> Result<(Daemon<OwnedRedbMessagePackBackingStore, WorkerSink>, CultNetRudpServerHub, CultNetSchemaRegistry)>
where
E: Embedder + Send + 'static,
V: VectorIndex + Send + 'static,
{
let mind = Mind::open(&options.state_root, &options.instance)?;
let hub = bind(options.bind, &runtime_id(&options.instance))?;
let startup_entries = mind.index_entries(None)?;
let sink = WorkerSink::spawn(embedder, index, &options.instance, mind.genesis_receipt_id()?, startup_entries, backoff);
Ok((Daemon::new(mind, sink), hub, schema_registry()?))
}
/// The bytes one reliable packet carries, and the packets one session may hold
@ -423,10 +456,14 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
mod tests {
use super::*;
use crate::daemon::tests::{
FITTING_CHANGES, FITTING_CUT, INSTANCE, OTHER, WIDE_CHANGES, WIDE_CUT, batch, identity, now, seeded_wide, slug,
FITTING_CHANGES, FITTING_CUT, INSTANCE, NoIndex, OTHER, WIDE_CHANGES, WIDE_CUT, batch, identity, now,
open_unindexed, seeded_wide, slug,
};
use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef};
use crate::envelope::{FAILURE_SCHEMA, decode_response, encode_request};
use crate::index::Backoff;
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use cultnet_rs::{CultMesh, CultMeshCdnArtifactManifest, CultMeshRudpSocketOptions, fetch_content};
use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, MindStatus};
use cultnet_rs::{FieldPredicate, Selection};
@ -476,7 +513,7 @@ mod tests {
#[test]
fn a_malformed_envelope_is_answered_with_a_failure_and_touches_no_mind() {
let root = tempfile::tempdir().unwrap();
let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
let registry = schema_registry().unwrap();
let admit = STANDARD.encode(
rmp_serde::to_vec_named(&HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)]))).unwrap(),
@ -557,7 +594,7 @@ mod tests {
#[test]
fn two_clients_get_their_own_replies_over_loopback() {
let root = tempfile::tempdir().unwrap();
let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
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();
@ -1104,7 +1141,7 @@ mod tests {
#[test]
fn an_overlong_echoed_field_is_refused_at_the_door_with_an_answer_that_fits() {
let root = tempfile::tempdir().unwrap();
let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let mut daemon = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
let registry = schema_registry().unwrap();
let mut bodies = fresh_bodies();
let huge = "x".repeat(1_300_000);
@ -1212,46 +1249,76 @@ mod tests {
#[test]
fn startup_opens_the_mind_before_it_binds_anything() {
let root = tempfile::tempdir().unwrap();
let held = Daemon::open(root.path(), &slug(INSTANCE)).unwrap();
let held = open_unindexed(root.path(), &slug(INSTANCE)).unwrap();
let occupied = UdpSocket::bind("127.0.0.1:0").unwrap();
let port = occupied.local_addr().unwrap();
let options = Options { state_root: root.path().to_path_buf(), instance: slug(INSTANCE), bind: port };
let error = format!("{:#}", startup(&options).err().expect("a held mind refuses"));
let options = Options {
state_root: root.path().to_path_buf(),
instance: slug(INSTANCE),
bind: port,
qdrant_url: "http://127.0.0.1:1".into(),
ollama_url: "http://127.0.0.1:1".into(),
embedding_model: "a-model".into(),
};
let started = |options: &Options| {
let (embedder, index) = crate::index::fakes::pair();
startup(options, embedder, index, Backoff::default())
};
let error = format!("{:#}", started(&options).err().expect("a held mind refuses"));
assert!(error.contains("MindAlreadyOwned"), "the refusal is the mind's, not the socket's: {error}");
assert!(!error.contains("binding"), "the socket was never reached: {error}");
// With the mind free and the port still held, the socket is the only
// gate left, and startup reports it.
drop(held);
let error = format!("{:#}", startup(&options).err().expect("a held port refuses"));
let error = format!("{:#}", started(&options).err().expect("a held port refuses"));
assert!(error.contains("binding"), "{error}");
drop(occupied);
let (_daemon, hub, _registry) = startup(&options).expect("both gates open");
let (_daemon, hub, _registry) = started(&options).expect("both gates open");
assert_eq!(hub.local_addr().unwrap(), port);
}
#[test]
fn options_are_exactly_state_root_instance_and_bind() {
fn options_are_exactly_the_six_the_operator_supplies() {
let root = if cfg!(windows) { r"C:\state" } else { "/state" };
let args = |values: &[&str]| values.iter().map(|value| value.to_string()).collect::<Vec<_>>().into_iter();
let good = [
"--state-root", root, "--instance", "yggdrasil", "--bind", "127.0.0.1:17872", "--qdrant-url",
"http://127.0.0.1:6333", "--ollama-url", "http://10.77.0.4:11434", "--embedding-model", "qwen3-embedding:0.6b",
];
assert_eq!(
parse_options(args(&["--state-root", root, "--instance", "yggdrasil", "--bind", "127.0.0.1:17872"]))
.unwrap(),
parse_options(args(&good)).unwrap(),
Options {
state_root: PathBuf::from(root),
instance: slug("yggdrasil"),
bind: "127.0.0.1:17872".parse().unwrap(),
qdrant_url: "http://127.0.0.1:6333".into(),
ollama_url: "http://10.77.0.4:11434".into(),
embedding_model: "qwen3-embedding:0.6b".into(),
}
);
// Each of the six is required: there is no index-less mode.
for name in ["--state-root", "--instance", "--bind", "--qdrant-url", "--ollama-url", "--embedding-model"] {
let at = good.iter().position(|value| *value == name).unwrap();
let mut without = good.to_vec();
without.drain(at..at + 2);
assert!(parse_options(args(&without)).is_err(), "{name} is required");
}
fn with<'a>(good: &[&'a str], name: &str, value: &'a str) -> Vec<&'a str> {
let at = good.iter().position(|option| *option == name).unwrap();
let mut changed = good.to_vec();
changed[at + 1] = value;
changed
}
for bad in [
vec!["--state-root", root, "--instance", "yggdrasil"],
vec!["--state-root", root, "--bind", "127.0.0.1:1"],
vec!["--instance", "yggdrasil", "--bind", "127.0.0.1:1"],
vec!["--state-root", root, "--instance", "yggdrasil", "--bind", "127.0.0.1:1", "--bind", "127.0.0.1:2"],
vec!["--state-root", root, "--instance", "yggdrasil", "--bind", "not-an-address"],
vec!["--state-root", "relative", "--instance", "yggdrasil", "--bind", "127.0.0.1:1"],
vec!["--state-root", root, "--instance", "yggdrasil", "--bind", "127.0.0.1:1", "--idunn-anchor", "x"],
[good.to_vec(), vec!["--bind", "127.0.0.1:2"]].concat(),
with(&good, "--bind", "not-an-address"),
with(&good, "--state-root", "relative"),
with(&good, "--qdrant-url", "https://127.0.0.1:6333"),
with(&good, "--ollama-url", "10.77.0.4:11434"),
with(&good, "--embedding-model", ""),
[good.to_vec(), vec!["--idunn-anchor", "x"]].concat(),
] {
assert!(parse_options(args(&bad)).is_err(), "{bad:?}");
}

View file

@ -0,0 +1,112 @@
//! A live check of the two adapters against a real Qdrant and a real Ollama.
//! Ignored by default: it needs both services, and it writes to Qdrant, so it
//! touches one scratch collection it creates (`huginn-smoke-<random>`, and a
//! dotted sibling to learn whether Qdrant accepts a `.` in a name) and deletes
//! both at the end. It never names, lists or reads any other collection.
//!
//! HUGINN_SMOKE_QDRANT=http://127.0.0.1:6333 \
//! HUGINN_SMOKE_OLLAMA=http://10.77.0.4:11434 \
//! HUGINN_SMOKE_MODEL=qwen3-embedding:0.6b \
//! cargo test -p huginn-daemon --test live_index_smoke -- --ignored --nocapture
use std::collections::BTreeSet;
use huginn_daemon::index::ollama::OllamaEmbedder;
use huginn_daemon::index::qdrant::QdrantIndex;
use huginn_daemon::index::{
CollectionMeta, Described, Embedder, MANAGED_BY, Point, PointPayload, VectorIndex, point_id,
};
use huginn_mind::INDEX_TEXT_VERSION;
fn env(name: &str) -> String {
std::env::var(name).unwrap_or_else(|_| panic!("{name} is required"))
}
fn raw(method: &str, url: &str, body: Option<&str>) -> (u16, String) {
let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
let mut response = match (method, body) {
("DELETE", _) => agent.delete(url).call(),
("POST", Some(body)) => agent.post(url).header("content-type", "application/json").send(body),
_ => agent.get(url).call(),
}
.expect("the request was sent");
let status = response.status().as_u16();
(status, response.body_mut().read_to_string().unwrap())
}
#[test]
#[ignore = "needs a live Qdrant and Ollama, and writes a scratch collection"]
fn the_adapters_embed_write_query_and_clean_up_a_scratch_collection() {
let (qdrant, ollama, model) = (env("HUGINN_SMOKE_QDRANT"), env("HUGINN_SMOKE_OLLAMA"), env("HUGINN_SMOKE_MODEL"));
let stamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos();
let plain = format!("huginn-smoke-{:x}-{stamp:x}", std::process::id());
let dotted = format!("{plain}.dotted");
let mut embedder = OllamaEmbedder::new(&ollama, &model);
let identity = embedder.model_identity().expect("the model is listed");
println!("identity: {identity:?}");
let texts = ["The organ owns the mind.".to_string(), "A recipe for lentil soup.".to_string()];
let vectors = embedder.embed(&texts).expect("two texts embed");
assert_eq!(vectors.len(), 2);
assert!(vectors.iter().all(|vector| vector.len() == identity.dimensions as usize));
let meta = CollectionMeta {
managed_by: MANAGED_BY.into(),
instance: "smoke".into(),
mind: "mind-commit-smoke".into(),
model: identity.name.clone(),
model_digest: identity.digest.clone(),
dimensions: identity.dimensions,
index_text_version: INDEX_TEXT_VERSION,
};
let mut store = QdrantIndex::new(&qdrant);
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
for name in [&plain, &dotted] {
assert_eq!(store.describe(name).unwrap(), Described::Absent);
store.recreate(name, &meta).unwrap();
assert_eq!(store.describe(name).unwrap(), Described::Labelled(meta.clone()));
assert!(store.ids(name).unwrap().is_empty());
let points: Vec<Point> = texts
.iter()
.zip(&vectors)
.enumerate()
.map(|(at, (text, vector))| Point {
id: point_id(&format!("smoke:doc:{at}")),
vector: vector.clone(),
payload: PointPayload {
doc_id: format!("smoke:doc:{at}"),
kind: "question".into(),
root: "smoke".into(),
ordinal: at as u64 + 1,
text_sha256: format!("{}", text.len()),
},
})
.collect();
store.upsert(name, &points).unwrap();
let expected: BTreeSet<String> = points.iter().map(|point| point.id.clone()).collect();
assert_eq!(store.ids(name).unwrap(), expected);
// The nearest point to the first text is the first point.
let query = serde_json_free_query(&vectors[0]);
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());
}
}));
for name in [&plain, &dotted] {
let (status, _) = raw("DELETE", &format!("{qdrant}/collections/{name}?wait=true"), None);
println!("deleted {name}: {status}");
}
let (status, _) = raw("GET", &format!("{qdrant}/collections/{plain}"), None);
assert_eq!(status, 404, "the scratch collection is gone");
if let Err(panic) = outcome {
std::panic::resume_unwind(panic);
}
}
/// The nearest-neighbour body, written out by hand so the test needs no JSON
/// crate of its own.
fn serde_json_free_query(vector: &[f32]) -> String {
let numbers: Vec<String> = vector.iter().map(|value| value.to_string()).collect();
format!("{{\"vector\":[{}],\"limit\":1}}", numbers.join(","))
}

View file

@ -680,6 +680,19 @@ mod tests {
assert!(Mind::open_with(store, &slug(OTHER_INSTANCE)).is_err());
}
/// The identity the index labels its collection with: none before the
/// first write, then the first receipt's id and never a later one.
#[test]
fn the_genesis_receipt_id_is_the_first_admissions_and_stays_put() {
let mut mind = opened(MemoryStore::new(), INSTANCE);
assert_eq!(mind.genesis_receipt_id().unwrap(), None);
let (first, _) = committed(admit(&mut mind, vec![instance(INSTANCE), stewardship(INSTANCE, REPO)]));
assert_eq!(mind.genesis_receipt_id().unwrap(), Some(first.clone()));
let (second, _) = committed(admit(&mut mind, vec![campaign(&[REPO])]));
assert_ne!(first, second);
assert_eq!(mind.genesis_receipt_id().unwrap(), Some(first));
}
#[test]
fn admission_refuses_a_foreign_instance_whatever_the_transport() {
let store = MemoryStore::new();

View file

@ -0,0 +1,312 @@
//! What of a document is searchable, and nothing else about the index.
//!
//! `index_text` is a rule about the leaf's documents: which fields are prose a
//! reader would search for. It is a pure function of one document, so it reads
//! no receipt, no provenance and no status, and nothing derived from it is
//! stored: the daemon's projection recomputes it, and `INDEX_TEXT_VERSION` is
//! how a change to it makes every projection rebuild. No network, no JSON, no
//! store handle: the daemon owns the vector store and the embedder.
use epiphany_pipeline::{PipelineDocument, PipelineKind, PipelineRef};
use sha2::{Digest, Sha256};
use crate::docs::Docs;
use crate::mind::Mind;
use crate::query::AdmissionIndex;
use crate::refusal::MindRefusal;
use crate::rows::root_and_local;
use crate::store::MindStore;
/// Bumped whenever `index_text` changes what it returns for any document.
pub const INDEX_TEXT_VERSION: u32 = 1;
/// The most bytes one document's text may carry, cut at a character boundary.
pub const INDEX_TEXT_MAX_BYTES: usize = 16 * 1024;
/// One document as the projection needs it: the key, the searchable text and
/// the digest of that text, and the ordinal of the receipt that wrote it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct IndexEntry {
pub id: PipelineRef,
/// The key's root segment: a campaign slug, or an instance.
pub root: String,
pub text: String,
pub text_sha256: String,
pub ordinal: u64,
}
/// The searchable text of a document, or `None` when its kind carries none
/// (`instance`, `stewardship`) or every field it names is empty. Lines are
/// joined by newlines. Withdrawn resolutions are indexed like any other: a
/// closed thing stays findable.
pub fn index_text(document: &PipelineDocument) -> Option<String> {
use PipelineDocument as D;
let mut lines: Vec<&str> = Vec::new();
match document {
D::Campaign(campaign) => lines.push(&campaign.title.0),
D::Target(target) => {
lines.extend(target.invariants.iter().map(|invariant| invariant.statement.0.as_str()));
lines.extend(target.not_in_scope.iter().map(|line| line.0.as_str()));
lines.extend(target.canonical_implementations.iter().map(|line| line.0.as_str()));
}
D::Question(question) => {
lines.push(&question.title.0);
lines.push(&question.question.0);
lines.extend(question.options.iter().map(|option| option.text.0.as_str()));
}
D::Ruling(ruling) => {
lines.push(&ruling.title.0);
lines.push(&ruling.ruling.0);
lines.extend(ruling.operator_quote.iter().map(|quote| quote.0.as_str()));
}
D::CutSpec(spec) => {
lines.push(&spec.title.0);
lines.extend(spec.first.iter().map(|line| line.0.as_str()));
lines.extend(spec.adds.iter().map(|line| line.0.as_str()));
lines.extend(spec.keeps_moves.iter().map(|line| line.0.as_str()));
}
D::CutReport(report) => {
lines.extend(report.commits.iter().map(|commit| commit.subject.0.as_str()));
for deviation in &report.deviations {
lines.push(&deviation.what.0);
lines.push(&deviation.why.0);
}
lines.extend(report.undone.iter().map(|line| line.0.as_str()));
lines.extend(report.promises.iter().map(|promise| promise.text.0.as_str()));
}
D::Verdict(verdict) => lines.extend(verdict.claims.iter().map(|claim| claim.claim.0.as_str())),
D::Finding(finding) => {
lines.push(&finding.claim.0);
lines.push(&finding.failure_scenario.0);
lines.extend(finding.evidence.iter().map(|evidence| evidence.result.0.as_str()));
}
D::FollowUp(follow_up) => {
lines.push(&follow_up.item.0);
lines.push(&follow_up.why_it_can_wait.0);
}
D::Resolution(resolution) => lines.push(&resolution.rationale.0),
D::HandOff(hand_off) => lines.push(&hand_off.reason.0),
D::Instance(_) | D::Stewardship(_) => {}
}
lines.retain(|line| !line.is_empty());
if lines.is_empty() {
return None;
}
let mut text = lines.join("\n");
if text.len() > INDEX_TEXT_MAX_BYTES {
let mut cut = INDEX_TEXT_MAX_BYTES;
while !text.is_char_boundary(cut) {
cut -= 1;
}
text.truncate(cut);
}
Some(text)
}
fn sha256_hex(text: &str) -> String {
Sha256::digest(text.as_bytes()).iter().map(|byte| format!("{byte:02x}")).collect()
}
impl<S: MindStore> Mind<S> {
/// The entries to project: every indexable document the mind holds, or
/// only the named ones. The ordinal is the writing receipt's. A named
/// document the mind does not hold is a refusal, not a silent omission: a
/// caller names what a commit just landed.
pub fn index_entries(&self, only: Option<&[PipelineRef]>) -> Result<Vec<IndexEntry>, MindRefusal> {
let documents: Vec<(PipelineKind, String, PipelineDocument)> = match only {
Some(refs) => {
let mut held = Vec::with_capacity(refs.len());
for reference in refs {
reference.validate_ref()?;
let document = self.get(reference.kind, &reference.id.0)?.ok_or_else(|| {
MindRefusal::Unavailable { detail: format!("index: {} is not held", reference.id.0) }
})?;
held.push((reference.kind, reference.id.0.clone(), document));
}
held
}
None => Docs::from_image(self.envelopes())?
.image
.into_iter()
.map(|held| (held.kind, held.key, held.document))
.collect(),
};
let facts = AdmissionIndex::build(self)?;
let mut entries = Vec::new();
for (kind, key, document) in documents {
let Some(text) = index_text(&document) else { continue };
let ordinal = facts.ordinal_of(kind, &key)?;
entries.push(IndexEntry {
root: root_and_local(&key).0.to_string(),
id: PipelineRef { kind, id: key.as_str().into() },
text_sha256: sha256_hex(&text),
text,
ordinal,
});
}
Ok(entries)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::fixtures::*;
use crate::receipt::Faculty;
use crate::store::test_stores::MemoryStore;
use epiphany_pipeline::{
ClaimOutcome, Deviation, FindingConfidence, PipelineDocument as D, PipelineKind as K, RulingAuthority,
};
use std::collections::BTreeMap;
/// A mind holding one document of every kind that carries text, each with
/// more than one text field where the kind has more than one, and the
/// question's closure withdrawn beside them. Returns the hand-off's key.
fn full_mind() -> (Mind<MemoryStore>, String) {
let mut mind = opened(MemoryStore::new(), INSTANCE);
let D::Target(mut target_doc) = target(1, &[INVARIANT, "second"]) else { panic!() };
target_doc.not_in_scope = vec!["No client writes.".into()];
target_doc.canonical_implementations = vec!["huginn-mind admission".into()];
committed(admit(
&mut mind,
vec![instance(INSTANCE), stewardship(INSTANCE, REPO), campaign(&[REPO]), D::Target(target_doc)],
));
committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A")]));
let mut ruling_doc = ruling("R1");
ruling_doc.answers = Some(s(&id("question", "Q1")));
ruling_doc.choice = Some(l("A"));
ruling_doc.operator_quote = Some("Own it.".into());
ruling_doc.authority = RulingAuthority::Operator;
committed(admit(&mut mind, vec![D::Ruling(ruling_doc)]));
// The derived closure of Q1 is withdrawn: a withdrawn resolution is a
// document like any other and stays findable.
let closed = id("resolution", "question.Q1.n1");
committed(admit(&mut mind, vec![resolution(r(K::Resolution, &closed), withdrawn())]));
let mut spec_doc = cut_spec("1", 1);
spec_doc.first = vec!["Read the leaf.".into()];
spec_doc.adds = vec!["An index module.".into(), "A worker.".into()];
spec_doc.keeps_moves = vec!["Admission stays.".into()];
committed(admit(&mut mind, vec![D::CutSpec(spec_doc)]));
let mut report_doc = cut_report("1", 1);
report_doc.deviations = vec![Deviation { what: "Changed a name.".into(), why: "It collided.".into() }];
report_doc.undone = vec!["The live smoke.".into()];
committed(admit(
&mut mind,
vec![D::CutReport(report_doc), follow_up("FU-1", r(K::Question, &id("question", "Q1")))],
));
committed(admit(
&mut mind,
vec![
verdict("1", 1, vec![claim(ClaimOutcome::Falsified, &[&id("finding", "cut-1.s1.F1")], Some("P1"), &["M1"])]),
D::Finding(finding("1", 1, "F1", FindingConfidence::Confirmed)),
],
));
let (_, writes) = committed(admit(&mut mind, vec![hand_off(INSTANCE, OTHER_INSTANCE, REPO, &[])]));
let hand_off_key = writes.iter().find(|write| write.kind == K::HandOff).unwrap().id.0.clone();
(mind, hand_off_key)
}
fn texts(mind: &Mind<MemoryStore>) -> BTreeMap<String, String> {
mind.index_entries(None).unwrap().into_iter().map(|entry| (entry.id.id.0, entry.text)).collect()
}
/// Each kind's search text is its named fields, joined by newlines, and
/// the two kinds with no prose have no entry.
#[test]
fn index_text_is_each_kinds_named_fields_and_the_identity_kinds_have_none() {
let (mind, hand_off_key) = full_mind();
let expected: Vec<(String, &str)> = vec![
(id("campaign", "self"), "Eureka pipeline state"),
(id("target", "r1"), "Only admission writes.\nOnly admission writes.\nNo client writes.\nhuginn-mind admission"),
(id("question", "Q1"), "Who owns the state?\nWho owns the state?\nan option\nan option"),
(id("ruling", "R1"), "An instance owns its mind\nAn instance owns its mind.\nOwn it."),
(id("cut_spec", "cut-1.r1"), "A cut\nRead the leaf.\nAn index module.\nA worker.\nAdmission stays."),
(
id("cut_report", "cut-1.h1"),
"Land the cut\nChanged a name.\nIt collided.\nThe live smoke.\nOne derived key per document.",
),
(id("verdict", "cut-1.s1"), "Falsified claim"),
(id("finding", "cut-1.s1.F1"), "A dotted label composes two keys.\nTwo documents claim one key.\nok"),
(id("follow_up", "FU-1"), "Per-kind admission rules.\nThe organ owns admission."),
// The withdrawn closure of Q1, admitted by hand.
(id("resolution", "resolution.question.Q1.n1.n1"), "Resolved."),
(hand_off_key, "The workstation mind takes the campaign."),
];
let held = texts(&mind);
for (key, text) in &expected {
match held.get(key) {
Some(found) => assert_eq!(found, text, "{key}"),
None => panic!("{key} has no entry; entries are {:?}", held.keys().collect::<Vec<_>>()),
}
}
let kinds: Vec<K> = mind.index_entries(None).unwrap().iter().map(|entry| entry.id.kind).collect();
assert!(!kinds.contains(&K::Instance) && !kinds.contains(&K::Stewardship), "{kinds:?}");
assert!(kinds.contains(&K::Resolution), "the derived closures are indexed too");
}
/// The text is the document's alone: the same document admitted by another
/// faculty at another clock, or closed by a later ruling, has the same
/// text and digest. Only the ordinal is the receipt's.
#[test]
fn index_text_ignores_admission_facts_and_indexes_withdrawn_resolutions() {
let mut mind = seeded();
committed(admit(&mut mind, vec![question("Q1", &["A", "B"], "A")]));
let q1 = id("question", "Q1");
let before = mind.index_entries(Some(&[r(K::Question, &q1)])).unwrap();
let mut other = seeded();
committed(admit_as(&mut other, Faculty::Soul, vec![question("Q1", &["A", "B"], "A")]));
let mut later = ruling("R1");
later.answers = Some(s(&q1));
later.choice = Some(l("A"));
committed(admit(&mut other, vec![D::Ruling(later)]));
let after = other.index_entries(Some(&[r(K::Question, &q1)])).unwrap();
assert_eq!(before, after, "a closed question is the same entry as an open one");
assert_eq!(before[0].ordinal, 2);
let closure = id("resolution", "question.Q1.n1");
committed(admit(&mut other, vec![resolution(r(K::Resolution, &closure), withdrawn())]));
let withdrawn_id = id("resolution", "resolution.question.Q1.n1.n1");
let entry = other.index_entries(Some(&[r(K::Resolution, &withdrawn_id)])).unwrap();
assert_eq!(entry.len(), 1, "a withdrawn resolution is indexed");
assert_eq!(entry[0].text, "Resolved.");
assert_eq!(entry[0].root, CAMPAIGN);
assert_eq!(entry[0].ordinal, 4);
}
#[test]
fn the_digest_is_sha256_of_the_text_and_a_missing_document_is_refused() {
let mind = seeded();
let campaign_ref = r(K::Campaign, &id("campaign", "self"));
let entry = &mind.index_entries(Some(std::slice::from_ref(&campaign_ref))).unwrap()[0];
assert_eq!(entry.text_sha256, "49c7e65adaad4bd34261300e9c5575847357b8f269508c9b2a46ec2aa75cc019");
assert_eq!((entry.root.as_str(), entry.ordinal), (CAMPAIGN, 1));
let absent = r(K::Question, &id("question", "Q9"));
assert!(matches!(mind.index_entries(Some(&[absent])), Err(MindRefusal::Unavailable { .. })));
// The identity and the stewardship carry no text: the seeded mind's
// indexable documents are its campaign and its target.
let all = mind.index_entries(None).unwrap();
assert_eq!(all.len(), 2, "{all:?}");
}
/// The bound is bytes, cut at a character boundary: three-byte characters
/// put the byte limit inside one.
#[test]
fn text_is_bounded_at_a_character_boundary() {
let wide: String = format!("{}x", "\u{20ac}".repeat(333));
assert_eq!(wide.len(), 1000);
let mut spec = cut_spec("1", 1);
spec.adds = (0..40).map(|_| wide.as_str().into()).collect();
let text = index_text(&D::CutSpec(spec)).unwrap();
// The title and its newline, then full lines of 1,001 bytes, then part
// of the next one, cut back to the last whole character.
let head = "A cut\n".len();
let stride = 1001;
let full_lines = (INDEX_TEXT_MAX_BYTES - head) / stride;
let consumed = head + full_lines * stride;
let inside = INDEX_TEXT_MAX_BYTES - consumed;
assert_ne!(inside % 3, 0, "the limit falls inside a character");
assert_eq!(text.len(), consumed + inside - inside % 3);
assert!(text.ends_with('\u{20ac}'));
}
}

View file

@ -26,6 +26,7 @@
pub mod admission;
mod docs;
pub mod index;
pub mod mind;
mod rows;
pub mod query;
@ -38,6 +39,7 @@ pub mod wire;
pub(crate) mod fixtures;
pub use docs::CitationRole;
pub use index::{INDEX_TEXT_MAX_BYTES, INDEX_TEXT_VERSION, IndexEntry, index_text};
pub use admission::{BATCH_MAX, PipelineAdmissionBatch, PipelineAdmissionOutcome};
pub use mind::{HuginnMindEpoch, Mind};
pub use query::{
@ -48,8 +50,8 @@ pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, PipelineProvena
pub use refusal::MindRefusal;
pub use store::{MindStore, OwnedRedbMessagePackBackingStore};
pub use wire::{
DeferredAnswer, HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA,
MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus,
DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON,
MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus,
};
/// The leaf, whole, through the one crate that pins its rev: the daemon and

View file

@ -245,6 +245,16 @@ impl<S: MindStore> Mind<S> {
.collect()
}
/// The mind's own identity: the receipt id of its first admission
/// (ordinal 1), which admission never rewrites and which digests the
/// instance and the first batch's bytes. `None` for a mind that has not
/// been written yet. The instance name is only a label, since two stores
/// on two hosts may share it; this tells them apart unless their first
/// batches were byte-identical.
pub fn genesis_receipt_id(&self) -> Result<Option<String>, MindRefusal> {
Ok(self.receipts()?.into_iter().find(|receipt| receipt.ordinal == 1).map(|receipt| receipt.receipt_id))
}
pub(crate) fn raw_envelope(&self, type_id: &str, key: &str) -> Option<&CultCacheEnvelope> {
self.image.iter().find(|envelope| envelope.r#type == type_id && envelope.key == key)
}

View file

@ -16,8 +16,8 @@
//! 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 11 wires the
//! index. No clock, no store handle, no network.
//! `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.
use std::collections::BTreeMap;
@ -304,11 +304,21 @@ impl AdmissionIndex {
}
fn of(&self, held: &Held) -> Result<AdmissionFacts, MindRefusal> {
let type_id = held.kind.type_id();
self.facts_of(held.kind, &held.key)
}
fn facts_of(&self, kind: PipelineKind, key: &str) -> Result<AdmissionFacts, MindRefusal> {
let type_id = kind.type_id();
self.0
.get(&(type_id.to_string(), held.key.clone()))
.get(&(type_id.to_string(), key.to_string()))
.cloned()
.ok_or_else(|| integrity(type_id, &held.key, "has no commit receipt"))
.ok_or_else(|| integrity(type_id, key, "has no commit receipt"))
}
/// The ordinal of the receipt that wrote a document: the projection's
/// position for it.
pub(crate) fn ordinal_of(&self, kind: PipelineKind, key: &str) -> Result<u64, MindRefusal> {
self.facts_of(kind, key).map(|facts| facts.ordinal)
}
}

View file

@ -135,6 +135,26 @@ impl HuginnMindResponse {
}
}
/// How far the semantic index has caught up with the mind. The mind decides
/// none of it: the daemon's index worker reports it, in memory, and a restart
/// recomputes it. `pending` counts documents not yet known to be in the index.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub enum IndexStatus {
/// The worker has not yet compared the mind with the collection.
Reconciling { pending: u32 },
/// Every indexable document is in the collection.
Current,
/// Reconciled, and indexing what is still pending.
Behind { pending: u32 },
/// The last attempt failed (the embedder or the vector store is
/// unreachable, or refused); it is retried with a backoff and nothing is
/// dropped. `attempts` counts consecutive failures.
Failing { pending: u32, attempts: u32, error: String },
/// The collection belongs to another writer. The worker never recreates
/// it; it re-reads it on the same backoff.
Refused { pending: u32, attempts: u32, reason: String },
}
/// What a mind says about itself: the typed state a dashboard projects.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct MindStatus {
@ -144,12 +164,13 @@ pub struct MindStatus {
/// not documents and are not counted here.
pub documents: u32,
pub receipts: u32,
pub index: IndexStatus,
}
impl<S: MindStore> Mind<S> {
/// `whoami`, derived from the image every time it is asked. Nothing stores
/// a status and no configuration supplies one.
pub fn status(&self) -> MindStatus {
/// `whoami`, derived from the image every time it is asked, and from the
/// index status the caller holds: the mind stores neither.
pub fn status(&self, index: IndexStatus) -> MindStatus {
let mut documents = 0_u32;
let mut receipts = 0_u32;
for envelope in self.envelopes() {
@ -164,6 +185,7 @@ impl<S: MindStore> Mind<S> {
schema_epoch: PIPELINE_SCHEMA_EPOCH.to_string(),
documents,
receipts,
index,
}
}
}
@ -192,7 +214,7 @@ mod tests {
fn responses() -> Vec<HuginnMindResponse> {
let refusal = MindRefusal::ForeignInstance { declared: OTHER_INSTANCE.into(), mind: INSTANCE.into() };
vec![
HuginnMindResponse::Whoami(seeded().status()),
HuginnMindResponse::Whoami(seeded().status(IndexStatus::Current)),
HuginnMindResponse::Admit(PipelineAdmissionOutcome::AlreadyAdmitted { receipt_id: "r1".into() }),
HuginnMindResponse::Admit(PipelineAdmissionOutcome::Conflict { identities: vec![] }),
HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(refusal.clone())),

View file

@ -469,6 +469,128 @@
"type": "string",
"maxLength": 40
},
"IndexStatus": {
"description": "How far the semantic index has caught up with the mind. The mind decides\nnone of it: the daemon's index worker reports it, in memory, and a restart\nrecomputes it. `pending` counts documents not yet known to be in the index.",
"oneOf": [
{
"description": "The worker has not yet compared the mind with the collection.",
"type": "object",
"properties": {
"Reconciling": {
"type": "object",
"properties": {
"pending": {
"type": "integer",
"format": "uint32",
"minimum": 0
}
},
"required": [
"pending"
]
}
},
"additionalProperties": false,
"required": [
"Reconciling"
]
},
{
"description": "Every indexable document is in the collection.",
"type": "string",
"const": "Current"
},
{
"description": "Reconciled, and indexing what is still pending.",
"type": "object",
"properties": {
"Behind": {
"type": "object",
"properties": {
"pending": {
"type": "integer",
"format": "uint32",
"minimum": 0
}
},
"required": [
"pending"
]
}
},
"additionalProperties": false,
"required": [
"Behind"
]
},
{
"description": "The last attempt failed (the embedder or the vector store is\nunreachable, or refused); it is retried with a backoff and nothing is\ndropped. `attempts` counts consecutive failures.",
"type": "object",
"properties": {
"Failing": {
"type": "object",
"properties": {
"attempts": {
"type": "integer",
"format": "uint32",
"minimum": 0
},
"error": {
"type": "string"
},
"pending": {
"type": "integer",
"format": "uint32",
"minimum": 0
}
},
"required": [
"pending",
"attempts",
"error"
]
}
},
"additionalProperties": false,
"required": [
"Failing"
]
},
{
"description": "The collection belongs to another writer. The worker never recreates\nit; it re-reads it on the same backoff.",
"type": "object",
"properties": {
"Refused": {
"type": "object",
"properties": {
"attempts": {
"type": "integer",
"format": "uint32",
"minimum": 0
},
"pending": {
"type": "integer",
"format": "uint32",
"minimum": 0
},
"reason": {
"type": "string"
}
},
"required": [
"pending",
"attempts",
"reason"
]
}
},
"additionalProperties": false,
"required": [
"Refused"
]
}
]
},
"Label": {
"description": "A key label: `[A-Za-z0-9_-]{1,64}`.",
"type": "string",
@ -1340,6 +1462,9 @@
"format": "uint32",
"minimum": 0
},
"index": {
"$ref": "#/$defs/IndexStatus"
},
"instance": {
"$ref": "#/$defs/Slug"
},
@ -1356,7 +1481,8 @@
"instance",
"schema_epoch",
"documents",
"receipts"
"receipts",
"index"
]
},
"MutationRecord": {