diff --git a/Cargo.lock b/Cargo.lock index 341aa15..3ecdc74 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/README.md b/README.md index 2317aa2..695c5ff 100644 --- a/README.md +++ b/README.md @@ -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_`, + 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. diff --git a/crates/huginn-daemon/Cargo.toml b/crates/huginn-daemon/Cargo.toml index 04dfd1a..428588e 100644 --- a/crates/huginn-daemon/Cargo.toml +++ b/crates/huginn-daemon/Cargo.toml @@ -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" diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index a1abc28..5742e61 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -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 { fn committed(&mut self, mind: &Mind, writes: &[PipelineRef]) -> anyhow::Result<()>; + fn status(&self) -> IndexStatus; } -/// This cut's index: there is none. -pub struct NoIndex; - -impl IndexSink for NoIndex { - fn committed(&mut self, _mind: &Mind, _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> { index: I, } -impl Daemon { - /// 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 { - Ok(Self { mind: Mind::open(state_root, instance)?, index: NoIndex }) - } -} - impl> Daemon { + /// 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, index: I) -> Self { + Self { mind, index } + } + pub fn with_index>(self, index: J) -> Daemon { Daemon { mind: self.mind, index } } @@ -51,9 +45,13 @@ impl> Daemon { &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> Daemon { 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) { 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) { 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 IndexSink for NoIndex { + fn committed(&mut self, _mind: &Mind, _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, MindRefusal> { + Ok(Daemon::new(Mind::open(state_root, instance)?, NoIndex)) + } + #[derive(Clone, Default)] struct RecordingIndex { seen: Arc, Vec)>>>, @@ -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, _writes: &[PipelineRef]) -> anyhow::Result<()> { Err(anyhow::anyhow!("the index is not there")) } + + fn status(&self) -> IndexStatus { + IndexStatus::Current + } } fn status(daemon: &mut Daemon>) -> 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::::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::::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() }) ); } diff --git a/crates/huginn-daemon/src/envelope.rs b/crates/huginn-daemon/src/envelope.rs index b3b0275..fecbde2 100644 --- a/crates/huginn-daemon/src/envelope.rs +++ b/crates/huginn-daemon/src/envelope.rs @@ -183,6 +183,7 @@ mod tests { schema_epoch: "epiphany.pipeline.epoch.v2".into(), documents: 0, receipts: 0, + index: huginn_mind::wire::IndexStatus::Current, }), "accepted", ), diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs new file mode 100644 index 0000000..c63041d --- /dev/null +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -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>, +} + +#[derive(Clone)] +pub(crate) struct FakeEmbedder { + pub(crate) state: Arc>, + gate: Option, +} + +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 { + self.state.lock().unwrap().embedded.iter().flatten().cloned().collect() + } +} + +impl Embedder for FakeEmbedder { + fn model_identity(&mut self) -> Result { + 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>> { + 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, +} + +#[derive(Default)] +pub(crate) struct IndexState { + pub(crate) collections: BTreeMap, + pub(crate) down: bool, + pub(crate) recreated: Vec<(String, CollectionMeta)>, + /// The point ids of every upsert call, in order. + pub(crate) upserts: Vec>, +} + +#[derive(Clone, Default)] +pub(crate) struct FakeIndex { + pub(crate) state: Arc>, +} + +impl FakeIndex { + pub(crate) fn holding(&self, collection: &str) -> BTreeSet { + 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 { + 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> { + 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()) +} diff --git a/crates/huginn-daemon/src/index/mod.rs b/crates/huginn-daemon/src/index/mod.rs new file mode 100644 index 0000000..ab864c1 --- /dev/null +++ b/crates/huginn-daemon/src/index/mod.rs @@ -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, + pub payload: PointPayload, +} + +pub trait Embedder { + fn model_identity(&mut self) -> Result; + /// One vector per text, in order. + fn embed(&mut self, texts: &[String]) -> Result>>; +} + +pub trait VectorIndex { + fn describe(&mut self, collection: &str) -> Result; + /// 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>; + 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 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 { + embedder: E, + index: V, + instance: String, + /// `None` until the mind's first admission; nothing is reconciled before. + mind: Option, + 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, + /// The ids in `known` whose points the collection may lack. + wanted: BTreeSet, + dimensions: Option, +} + +impl Projector { + pub fn new(embedder: E, index: V, instance: &Slug, mind: Option) -> 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) { + 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 { + let stepped = self.step(); + if stepped.is_err() { + self.dimensions = None; + } + stepped + } + + fn step(&mut self) -> Result { + 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 = + 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 = 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, + /// The mind's identity, once a commit has revealed it to a mind that had none. + mind: Option, + status: IndexStatus, + closed: bool, +} + +type Lock = Arc<(Mutex, 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) { + 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(mut projector: Projector, 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( + embedder: E, + index: V, + instance: &Slug, + mind: Option, + startup: Vec, + 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 IndexSink for WorkerSink { + fn committed(&mut self, mind: &Mind, 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; diff --git a/crates/huginn-daemon/src/index/ollama.rs b/crates/huginn-daemon/src/index/ollama.rs new file mode 100644 index 0000000..850b276 --- /dev/null +++ b/crates/huginn-daemon/src/index/ollama.rs @@ -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: \nQuery: `. +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 { + 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::()); + 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 { + 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>> { + 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() + } +} diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs new file mode 100644 index 0000000..a794ac1 --- /dev/null +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -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::()); + 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 { + 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> { + 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 = 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(()) + } +} diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs new file mode 100644 index 0000000..90524b8 --- /dev/null +++ b/crates/huginn-daemon/src/index/tests.rs @@ -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) { + 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) { + 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(projector: &mut Projector) { + while projector.advance().unwrap() == Advance::Progressed {} +} + +fn point_ids(entries: &[IndexEntry]) -> BTreeSet { + entries.iter().map(|entry| point_id(&entry.id.id.0)).collect() +} + +fn projected(embedder: &FakeEmbedder, index: &FakeIndex, entries: &[IndexEntry]) -> Projector { + 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 = 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 = 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); 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) -> 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>>, +} + +type Route = (&'static str, u16, &'static str); + +fn find(haystack: &[u8], needle: &[u8]) -> Option { + haystack.windows(needle.len()).position(|window| window == needle) +} + +/// Serves each route once, in order, then repeats the last route that matched. +/// A route matches `" "` 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>> = 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>, + log: &Mutex>, +) { + let mut buffer: Vec = 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::().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 = + 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 = ["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 = 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_`. +#[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()); +} diff --git a/crates/huginn-daemon/src/lib.rs b/crates/huginn-daemon/src/lib.rs index d2dc669..fc27e59 100644 --- a/crates/huginn-daemon/src/lib.rs +++ b/crates/huginn-daemon/src/lib.rs @@ -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}; diff --git a/crates/huginn-daemon/src/main.rs b/crates/huginn-daemon/src/main.rs index 2c14b8d..ce28fb3 100644 --- a/crates/huginn-daemon/src/main.rs +++ b/crates/huginn-daemon/src/main.rs @@ -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); diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index d7c1d63..287f7b9 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -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) -> Result { let mut args = args; let mut values: BTreeMap = 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) -> Result { 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 { + 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( options: &Options, -) -> Result<(Daemon, 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, 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>( 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::>().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:?}"); } diff --git a/crates/huginn-daemon/tests/live_index_smoke.rs b/crates/huginn-daemon/tests/live_index_smoke.rs new file mode 100644 index 0000000..8baaf69 --- /dev/null +++ b/crates/huginn-daemon/tests/live_index_smoke.rs @@ -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-`, 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 = 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 = 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 = vector.iter().map(|value| value.to_string()).collect(); + format!("{{\"vector\":[{}],\"limit\":1}}", numbers.join(",")) +} diff --git a/crates/huginn-mind/src/admission.rs b/crates/huginn-mind/src/admission.rs index 8ea8f8a..4b85cdd 100644 --- a/crates/huginn-mind/src/admission.rs +++ b/crates/huginn-mind/src/admission.rs @@ -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(); diff --git a/crates/huginn-mind/src/index.rs b/crates/huginn-mind/src/index.rs new file mode 100644 index 0000000..7ac4869 --- /dev/null +++ b/crates/huginn-mind/src/index.rs @@ -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 { + 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 Mind { + /// 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, 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, 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) -> BTreeMap { + 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::>()), + } + } + let kinds: Vec = 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}')); + } +} diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index 451e73f..8029548 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -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 diff --git a/crates/huginn-mind/src/mind.rs b/crates/huginn-mind/src/mind.rs index 8deed10..ff5fbc2 100644 --- a/crates/huginn-mind/src/mind.rs +++ b/crates/huginn-mind/src/mind.rs @@ -245,6 +245,16 @@ impl Mind { .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, 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) } diff --git a/crates/huginn-mind/src/query.rs b/crates/huginn-mind/src/query.rs index e657509..a85e353 100644 --- a/crates/huginn-mind/src/query.rs +++ b/crates/huginn-mind/src/query.rs @@ -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 { - let type_id = held.kind.type_id(); + self.facts_of(held.kind, &held.key) + } + + fn facts_of(&self, kind: PipelineKind, key: &str) -> Result { + 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 { + self.facts_of(kind, key).map(|facts| facts.ordinal) } } diff --git a/crates/huginn-mind/src/wire.rs b/crates/huginn-mind/src/wire.rs index 5f46950..dcbd491 100644 --- a/crates/huginn-mind/src/wire.rs +++ b/crates/huginn-mind/src/wire.rs @@ -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 Mind { - /// `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 Mind { schema_epoch: PIPELINE_SCHEMA_EPOCH.to_string(), documents, receipts, + index, } } } @@ -192,7 +214,7 @@ mod tests { fn responses() -> Vec { 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())), diff --git a/schemas/cultnet/huginn.mind_response.v1.schema.json b/schemas/cultnet/huginn.mind_response.v1.schema.json index 6164683..6f4cc21 100644 --- a/schemas/cultnet/huginn.mind_response.v1.schema.json +++ b/schemas/cultnet/huginn.mind_response.v1.schema.json @@ -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": {