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

chat: search_chat over the shared source collection (wire, payload contract, Qdrant, queue)

Browse source 
Probe: the deployed Qdrant answers 1.17.1 at 127.0.0.1:6333 (>= 1.8: integer
range payload indexes and scroll order_by exist).

Wire: SearchChat / Chat, ChatQuery (door-checked by field), ChatHit with no
optional room_id, event ids or timestamps, published schemas regenerated.
Index: corpus payload on every point (forge searches must corpus=forge), chat
payload codec derived from the messages, keyword and integer indexes, upsert_chat
/ chat_nearest / chat_scroll on Qdrant and the fakes. CORPUS becomes
forge:GameCult+chat:rooms, so the existing meta mismatch rebuilds the forge index
once at deploy. sources::search_chat shares the identity check and fuse with
the forge search; chat rides the same queue (SourceAsk) and Found::Chat.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Eureka 2026-11-11 17:34:21 +00:00
commit 3b8fc7da1f
14 changed files with 3139 additions and 2065 deletions
Download patch file Download diff file Expand all files Collapse all files

1
crates/eureka-state/src/bin/huginn/main.rs
View file

@ -688,6 +688,7 @@ fn unexpected(answer: &HuginnMindResponse) -> Trouble {
HuginnMindResponse::Persona(_) => "a persona",
HuginnMindResponse::PersonaPut(_) => "a persona put",
HuginnMindResponse::Sources(_) => "source hits",
HuginnMindResponse::Chat(_) => "chat hits",
HuginnMindResponse::Activity(_) => "repo activity",
HuginnMindResponse::RootsTouched(_) => "roots touched",
HuginnMindResponse::Refused(_) => "a refusal",

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

@ -1233,7 +1233,8 @@ struct Desk {
embeds_under: &'static str,
held: Vec<huginn_daemon::sources::SourcePayload>,
tickets: u64,
done: Vec<(SearchTicket, Result<huginn_mind::SourceHits, String>)>,
chat: Vec<huginn_daemon::sources::ChatPayload>,
done: Vec<(SearchTicket, huginn_daemon::daemon::Found)>,
}
/// An embedder that sends every text to one vector, under a digest.
@ -1297,6 +1298,35 @@ impl huginn_daemon::sources::SourceIndex for Desk {
.map(|payload| huginn_daemon::sources::SourceRow { score: 0.9, payload: payload.clone() })
.collect())
}
fn upsert_chat(&mut self, _points: &[huginn_daemon::sources::ChatPoint]) -> anyhow::Result<()> {
anyhow::bail!("a daemon never writes the source index")
}
fn chat_nearest(
&mut self,
_vector: &[f32],
filter: &huginn_daemon::sources::ChatFilter,
limit: u32,
_within: Duration,
) -> anyhow::Result<Vec<huginn_daemon::sources::ChatRow>> {
Ok(self.chat_scroll(filter, limit, _within)?.into_iter().map(|payload| huginn_daemon::sources::ChatRow { score: 0.9, payload }).collect())
}
fn chat_scroll(&mut self, filter: &huginn_daemon::sources::ChatFilter, limit: u32, _within: Duration) -> anyhow::Result<Vec<huginn_daemon::sources::ChatPayload>> {
let mut kept: Vec<_> = self
.chat
.iter()
.filter(|payload| {
let times = payload.messages.iter().map(|message| message.ts_ms);
filter.keeps(&payload.room_id, payload.room_alias.as_deref(), times.clone().min().unwrap(), times.max().unwrap())
})
.cloned()
.collect();
kept.sort_by_key(|payload| payload.messages[0].ts_ms);
kept.truncate(limit as usize);
Ok(kept)
}
}
impl<S: MindStore> IndexSink<S> for Desk {
@ -1313,11 +1343,21 @@ impl<S: MindStore> IndexSink<S> for Desk {
self.tickets += 1;
let ticket = SearchTicket(self.tickets);
let mut embedder = Flat(self.embeds_under);
let found = huginn_daemon::sources::search(&mut embedder, self, query, Duration::from_secs(5)).map_err(|error| format!("{error:#}"));
let found = huginn_daemon::sources::search(&mut embedder, self, query, Duration::from_secs(5))
.map_or_else(|error| huginn_daemon::daemon::Found::Failed(format!("{error:#}")), huginn_daemon::daemon::Found::Sources);
self.done.push((ticket, found));
Ok(ticket)
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<huginn_mind::SourceHits, String>)> {
fn search_chat(&mut self, query: &huginn_mind::ChatQuery) -> Result<SearchTicket, String> {
self.tickets += 1;
let ticket = SearchTicket(self.tickets);
let mut embedder = Flat(self.embeds_under);
let found = huginn_daemon::sources::search_chat(&mut embedder, self, query, Duration::from_secs(5))
.map_or_else(|error| huginn_daemon::daemon::Found::Failed(format!("{error:#}")), huginn_daemon::daemon::Found::Chat);
self.done.push((ticket, found));
Ok(ticket)
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, huginn_daemon::daemon::Found)> {
std::mem::take(&mut self.done)
}
fn search_deadline(&self) -> Duration {
@ -1351,6 +1391,7 @@ fn source_daemon(built_under: &'static str, embeds_under: &'static str) -> Serve
chunk("Demo", "README.md", 1, 2, "# Demo"),
chunk("Other", "x.rs", 1, 1, "fn other() {}"),
],
chat: Vec::new(),
tickets: 0,
done: Vec::new(),
};

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

@ -93,10 +93,14 @@ impl<S: MindStore> IndexSink<S> for Inert {
fn search(&mut self, _text: &str, _top_k: u32, _among: Vec<String>) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn search_chat(&mut self, _query: &huginn_mind::ChatQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn search_sources(&mut self, _query: &huginn_mind::SourceQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<huginn_mind::SourceHits, String>)> {
fn sources_searched(&mut self) -> Vec<(SearchTicket, huginn_daemon::daemon::Found)> {
Vec::new()
}
fn search_deadline(&self) -> Duration {

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

@ -8,7 +8,7 @@ use crate::sources::forge::HttpForge;
use huginn_mind::desk_wire::{DeskRequest, DeskResponse};
use huginn_mind::eureka_pipeline::{PipelineRef, Slug};
use huginn_mind::wire::{ACTIVITY_REPOS_MAX, HuginnMindRequest, HuginnMindResponse, IndexStatus};
use huginn_mind::{Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery, SourceHits, SourceQuery};
use huginn_mind::{ChatHits, ChatQuery, Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery, SourceHits, SourceQuery};
/// Where the index attaches. Called after a batch commits, with the refs the
/// batch landed, derived writes included; the sink may read each through
@ -34,8 +34,12 @@ pub trait IndexSink<S: MindStore> {
/// Starts a source search. The source index is not the mind's: it needs no
/// member ids, and nothing about the mind's own index state refuses it.
fn search_sources(&mut self, query: &SourceQuery) -> Result<SearchTicket, String>;
/// Every source search that finished since the last call, by ticket.
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)>;
/// Starts a chat search: the same queue and worker as a source search,
/// over the same collection.
fn search_chat(&mut self, query: &ChatQuery) -> Result<SearchTicket, String>;
/// Every source or chat search that finished since the last call, by
/// ticket: `Found::Sources`, `Found::Chat` or `Found::Failed`.
fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)>;
fn search_deadline(&self) -> std::time::Duration;
/// Every search that finished since the last call, by ticket: its
/// candidates and scores, or why it failed.
@ -78,6 +82,7 @@ struct MindSearch {
pub enum Found {
Mind(Hits),
Sources(SourceHits),
Chat(ChatHits),
Failed(String),
}
@ -205,6 +210,15 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
Ok(roots) => HuginnMindResponse::RootsTouched(roots),
Err(refusal) => HuginnMindResponse::Refused(refusal),
},
HuginnMindRequest::SearchChat { query, .. } => {
if let Err(refusal) = query.check() {
return Handled::Answered(HuginnMindResponse::Refused(refusal));
}
return match self.index.search_chat(&query) {
Ok(ticket) => Handled::Searching(Search { ticket, mind: None }),
Err(detail) => Handled::Answered(HuginnMindResponse::Refused(MindRefusal::Unavailable { detail })),
};
}
HuginnMindRequest::StewardedActivity { instance, repos, after } => {
let refused = |message: &str| MindRefusal::SelectionInvalid { field: "repos".into(), value: None, message: message.into() };
if repos.len() > ACTIVITY_REPOS_MAX {
@ -282,8 +296,7 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
/// Searches the index has finished since this was last called.
pub fn searched(&mut self) -> Vec<(SearchTicket, Found)> {
let mind = self.index.searched().into_iter().map(|(ticket, found)| (ticket, found.map_or_else(Found::Failed, Found::Mind)));
let sources = self.index.sources_searched().into_iter().map(|(ticket, found)| (ticket, found.map_or_else(Found::Failed, Found::Sources)));
mind.chain(sources).collect()
mind.chain(self.index.sources_searched()).collect()
}
/// The client has stopped waiting for `ticket`.
@ -300,6 +313,7 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
let (selection, semantic, members, hits) = match (search.mind, found) {
(_, Found::Failed(detail)) => return unavailable(detail),
(None, Found::Sources(hits)) => return HuginnMindResponse::Sources(hits),
(None, Found::Chat(hits)) => return HuginnMindResponse::Chat(hits),
(Some(asked), Found::Mind(hits)) => (asked.selection, asked.semantic, asked.members, hits),
_ => return unavailable("the index answered another kind of search than the one asked".into()),
};
@ -326,6 +340,7 @@ pub(crate) mod tests {
StructuralDelta, Title, VerificationTest,
};
use huginn_mind::wire::MindStatus;
use huginn_mind::ChatQuery;
use huginn_mind::{MindRefusal, OwnedRedbMessagePackBackingStore};
use std::path::Path;
use cultnet_rs::Selection;
@ -580,11 +595,15 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn search_chat(&mut self, _query: &ChatQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
Vec::new()
}
@ -619,13 +638,17 @@ pub(crate) mod tests {
Err("this index cannot search the mind".into())
}
fn search_chat(&mut self, _query: &ChatQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
*self.asked.lock().unwrap() += 1;
Ok(SearchTicket(7))
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
vec![(SearchTicket(7), Ok(self.hits.clone()))]
fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
vec![(SearchTicket(7), Found::Sources(self.hits.clone()))]
}
fn search_deadline(&self) -> std::time::Duration {
@ -685,6 +708,31 @@ pub(crate) mod tests {
assert_eq!(*asked.lock().unwrap(), 3, "a refused query never reached the index");
}
/// A chat search is checked at the door before the index is asked, is
/// refused for another mind, turns an index that cannot be asked into
/// `Unavailable`, and its finished hits are the answer as they are.
#[test]
fn a_chat_search_is_checked_at_the_door_and_answers_its_hits() {
use huginn_mind::eureka_pipeline::{Line, Short};
use huginn_mind::{ChatHits, SourceModel};
let root = tempfile::tempdir().unwrap();
let mut daemon = Daemon::new(Mind::open(root.path(), &slug(INSTANCE)).unwrap(), NoIndex);
let request = |instance: &str, rooms: &[&str], top_k: u32| HuginnMindRequest::SearchChat {
instance: slug(instance),
query: ChatQuery { texts: vec![Line("find".into())], rooms: rooms.iter().map(|room| Short((*room).into())).collect(), after_ms: None, before_ms: None, top_k },
};
for (refused, field) in [(request(INSTANCE, &[], 0), "top_k"), (request(INSTANCE, &["swarm"], 5), "rooms")] {
let answer = daemon.handle(refused, now()).answered();
assert!(matches!(answer, HuginnMindResponse::Refused(MindRefusal::SelectionInvalid { field: ref named, .. }) if named == field), "{answer:?}");
}
assert!(matches!(daemon.handle(request(OTHER, &[], 5), now()).answered(), HuginnMindResponse::Refused(_)));
let answer = daemon.handle(request(INSTANCE, &["#swarm:x"], 5), now()).answered();
assert!(matches!(answer, HuginnMindResponse::Refused(MindRefusal::Unavailable { .. })), "{answer:?}");
let hits = ChatHits { indexed_under: SourceModel { name: "m".into(), digest: "d".into() }, truncated: false, hits: vec![] };
let search = Search { ticket: SearchTicket(1), mind: None };
assert_eq!(daemon.finish(search, Found::Chat(hits.clone())), HuginnMindResponse::Chat(hits));
}
/// A daemon over the instance's mind under `state_root`, with no index.
pub(crate) fn open_unindexed(
state_root: &Path,
@ -716,11 +764,15 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn search_chat(&mut self, _query: &ChatQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
Vec::new()
}
@ -750,11 +802,15 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn search_chat(&mut self, _query: &ChatQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
Err("this index cannot search".into())
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
Vec::new()
}

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

@ -8,7 +8,7 @@ use std::time::Duration;
use anyhow::{Result, bail, ensure};
use crate::sources::{DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta};
use crate::sources::{ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta};
use super::{CollectionMeta, Described, Embedder, Hit, ModelIdentity, Point, VectorIndex, point_id};
@ -253,6 +253,8 @@ pub(crate) fn pair() -> (FakeEmbedder, FakeIndex) {
pub(crate) struct SourcesState {
pub(crate) described: Option<DescribedSources>,
pub(crate) points: BTreeMap<String, SourcePoint>,
/// The chat points, by id: the same collection, the other corpus.
pub(crate) chat: BTreeMap<String, ChatPoint>,
/// How many times the collection was recreated.
pub(crate) recreated: u32,
pub(crate) upsert_calls: u32,
@ -317,6 +319,7 @@ impl SourceIndex for FakeSources {
let mut state = self.state.lock().unwrap();
state.recreated += 1;
state.points.clear();
state.chat.clear();
state.described = Some(DescribedSources::Labelled { meta: meta.clone(), cursors: BTreeMap::new() });
Ok(())
}
@ -369,6 +372,44 @@ impl SourceIndex for FakeSources {
rows.truncate(limit as usize);
Ok(rows)
}
fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> {
let mut state = self.state.lock().unwrap();
for point in points {
ensure!(!point.payload.messages.is_empty(), "a chat chunk carries no messages");
state.chat.insert(point.id.clone(), point.clone());
}
Ok(())
}
fn chat_nearest(&mut self, vector: &[f32], filter: &ChatFilter, limit: u32, _within: Duration) -> Result<Vec<ChatRow>> {
let state = self.state.lock().unwrap();
let mut rows: Vec<ChatRow> = state
.chat
.values()
.filter(|point| chat_kept(filter, &point.payload))
.map(|point| ChatRow { score: cosine(vector, &point.vector), payload: point.payload.clone() })
.collect();
rows.sort_by(|a, b| b.score.total_cmp(&a.score));
rows.truncate(limit as usize);
Ok(rows)
}
fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, _within: Duration) -> Result<Vec<ChatPayload>> {
let state = self.state.lock().unwrap();
let mut kept: Vec<&ChatPayload> = state.chat.values().map(|point| &point.payload).filter(|payload| chat_kept(filter, payload)).collect();
kept.sort_by_key(|payload| payload.messages.iter().map(|message| message.ts_ms).min());
Ok(kept.into_iter().take(limit as usize).cloned().collect())
}
}
/// A chat chunk under a filter, by the span its messages make.
fn chat_kept(filter: &ChatFilter, payload: &ChatPayload) -> bool {
let times = || payload.messages.iter().map(|message| message.ts_ms);
match (times().min(), times().max()) {
(Some(start), Some(end)) => filter.keeps(&payload.room_id, payload.room_alias.as_deref(), start, end),
_ => false,
}
}
impl SourceIndex for FakeIndex {
@ -390,4 +431,13 @@ impl SourceIndex for FakeIndex {
fn search_sources(&mut self, vector: &[f32], repo: Option<&str>, front: bool, limit: u32, within: Duration) -> Result<Vec<SourceRow>> {
self.sources.search_sources(vector, repo, front, limit, within)
}
fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> {
self.sources.upsert_chat(points)
}
fn chat_nearest(&mut self, vector: &[f32], filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatRow>> {
self.sources.chat_nearest(vector, filter, limit, within)
}
fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatPayload>> {
self.sources.chat_scroll(filter, limit, within)
}
}

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

@ -105,9 +105,9 @@ use huginn_mind::wire::IndexStatus;
use huginn_mind::{INDEX_TEXT_VERSION, IndexEntry, Mind, MindStore};
use sha2::{Digest, Sha256};
use crate::daemon::{Hits, IndexSink, SearchTicket};
use crate::daemon::{Found, Hits, IndexSink, SearchTicket};
use crate::sources::{self, SourceIndex};
use huginn_mind::{SourceHits, SourceQuery};
use huginn_mind::{ChatQuery, SourceQuery};
use ollama::QUERY_INSTRUCTION;
/// The `managed_by` a collection carries when this organ made it.
@ -685,11 +685,17 @@ struct Queued {
due: Instant,
}
/// A source search waiting for the worker. It is not the mind's: nothing about
/// the mind's reconciliation gates it.
/// What a source-collection search asks: forge chunks or chat chunks.
enum SourceAsk {
Sources(SourceQuery),
Chat(ChatQuery),
}
/// A source or chat search waiting for the worker. It is not the mind's:
/// nothing about the mind's reconciliation gates it.
struct QueuedSource {
ticket: SearchTicket,
query: SourceQuery,
ask: SourceAsk,
due: Instant,
}
@ -706,7 +712,7 @@ struct Shared {
/// Searches the worker finished and the serve loop has not collected.
found: Vec<(SearchTicket, Result<Hits, String>)>,
source_searches: VecDeque<QueuedSource>,
sources_found: Vec<(SearchTicket, Result<SourceHits, String>)>,
sources_found: Vec<(SearchTicket, Found)>,
next_ticket: u64,
}
@ -821,10 +827,14 @@ fn serve_source_search<E: Embedder, V: VectorIndex + SourceIndex>(projector: &mu
if guard.closed {
return;
}
let Some(QueuedSource { ticket, query, due }) = guard.source_searches.pop_front() else { return };
let Some(QueuedSource { ticket, ask, due }) = guard.source_searches.pop_front() else { return };
drop(guard);
let found = sources::search(&mut projector.embedder, &mut projector.index, &query, due.saturating_duration_since(Instant::now()))
.map_err(|error| format!("{error:#}"));
let within = due.saturating_duration_since(Instant::now());
let found = match ask {
SourceAsk::Sources(query) => sources::search(&mut projector.embedder, &mut projector.index, &query, within).map(Found::Sources),
SourceAsk::Chat(query) => sources::search_chat(&mut projector.embedder, &mut projector.index, &query, within).map(Found::Chat),
}
.unwrap_or_else(|error| Found::Failed(format!("{error:#}")));
lock(shared).sources_found.push((ticket, found));
}
@ -966,6 +976,24 @@ pub struct WorkerSink {
}
impl WorkerSink {
/// Queues a source or chat search behind the same bound as every other
/// search; the mind's index state does not refuse it.
fn queue_source_ask(&mut self, ask: SourceAsk) -> Result<SearchTicket, String> {
let mut guard = lock(&self.shared);
if guard.searches.len() + guard.source_searches.len() >= SEARCHES_QUEUED_MAX {
return Err(format!(
"the index has {} searches waiting and takes no more until it has answered some",
guard.searches.len() + guard.source_searches.len()
));
}
let ticket = SearchTicket(guard.next_ticket);
guard.next_ticket += 1;
guard.source_searches.push_back(QueuedSource { ticket, ask, due: Instant::now() + self.deadline });
drop(guard);
self.shared.1.notify_all();
Ok(ticket)
}
/// 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
@ -1086,22 +1114,14 @@ impl<S: MindStore> IndexSink<S> for WorkerSink {
}
fn search_sources(&mut self, query: &SourceQuery) -> Result<SearchTicket, String> {
let mut guard = lock(&self.shared);
if guard.searches.len() + guard.source_searches.len() >= SEARCHES_QUEUED_MAX {
return Err(format!(
"the index has {} searches waiting and takes no more until it has answered some",
guard.searches.len() + guard.source_searches.len()
));
}
let ticket = SearchTicket(guard.next_ticket);
guard.next_ticket += 1;
guard.source_searches.push_back(QueuedSource { ticket, query: query.clone(), due: Instant::now() + self.deadline });
drop(guard);
self.shared.1.notify_all();
Ok(ticket)
self.queue_source_ask(SourceAsk::Sources(query.clone()))
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
fn search_chat(&mut self, query: &ChatQuery) -> Result<SearchTicket, String> {
self.queue_source_ask(SourceAsk::Chat(query.clone()))
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
std::mem::take(&mut lock(&self.shared).sources_found)
}

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

@ -7,7 +7,12 @@ use std::time::Duration;
use anyhow::{Context, Result, ensure};
use serde_json::{Value, json};
use crate::sources::{COLLECTION, FRONT_PAGE_FILES, DescribedSources, SourceIndex, SourcePayload, SourcePoint, SourceRow, SourcesMeta};
use huginn_mind::ChatMessage;
use crate::sources::{
COLLECTION, CORPUS_CHAT, CORPUS_FORGE, ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, FRONT_PAGE_FILES, SourceIndex, SourcePayload, SourcePoint, SourceRow,
SourcesMeta,
};
use super::{CollectionMeta, INDEX_CALL_TIMEOUT, Described, Hit, Http, Method, Point, VectorIndex};
@ -175,21 +180,94 @@ impl VectorIndex for QdrantIndex {
}
}
/// The payload filter of a source search: the repo, and the front-page paths
/// (exact matches on the keyword-indexed `path`), both when both are asked.
fn source_filter(repo: Option<&str>, front: bool) -> Option<Value> {
let mut must = Vec::new();
/// The payload filter of a source search: forge points only, then the repo and
/// the front-page paths (exact matches on the keyword-indexed `path`) when
/// asked. A chat point carries no repo or path and can never pass it.
fn source_filter(repo: Option<&str>, front: bool) -> Value {
let mut must = vec![json!({ "key": "corpus", "match": { "value": CORPUS_FORGE } })];
if let Some(repo) = repo {
must.push(json!({ "key": "repo", "match": { "value": repo } }));
}
if front {
must.push(json!({ "key": "path", "match": { "any": FRONT_PAGE_FILES } }));
}
(!must.is_empty()).then(|| json!({ "must": must }))
json!({ "must": must })
}
/// The payload filter of a chat read: chat points only; the rooms any-of by id
/// (`!`) or alias (`#`) as a `should`; the span as ranges that keep a chunk
/// overlapping [after, before].
fn chat_filter(filter: &ChatFilter) -> Value {
let mut must = vec![json!({ "key": "corpus", "match": { "value": CORPUS_CHAT } })];
if let Some(after) = filter.after_ms {
must.push(json!({ "key": "ts_end_ms", "range": { "gte": after } }));
}
if let Some(before) = filter.before_ms {
must.push(json!({ "key": "ts_start_ms", "range": { "lte": before } }));
}
let mut out = json!({ "must": must });
let (ids, aliases): (Vec<&String>, Vec<&String>) = filter.rooms.iter().partition(|room| room.starts_with('!'));
let mut should = Vec::new();
if !ids.is_empty() {
should.push(json!({ "key": "room_id", "match": { "any": ids } }));
}
if !aliases.is_empty() {
should.push(json!({ "key": "room_alias", "match": { "any": aliases } }));
}
if !should.is_empty() {
out["should"] = Value::Array(should);
}
out
}
/// A chat chunk's payload. The span, first and last ids and the id list are
/// derived from `messages`, so they cannot disagree with them.
fn chat_payload_json(payload: &ChatPayload) -> Result<Value> {
let (Some(first), Some(last)) = (payload.messages.first(), payload.messages.last()) else {
anyhow::bail!("a chat chunk of {} carries no messages", payload.room_id);
};
let mut out = json!({
"corpus": CORPUS_CHAT,
"room_id": payload.room_id,
"ts_start_ms": payload.messages.iter().map(|message| message.ts_ms).min(),
"ts_end_ms": payload.messages.iter().map(|message| message.ts_ms).max(),
"first_event_id": first.event_id,
"last_event_id": last.event_id,
"event_ids": payload.messages.iter().map(|message| &message.event_id).collect::<Vec<_>>(),
"messages": payload.messages.iter().map(|message| json!({
"event_id": message.event_id, "sender": message.sender, "ts_ms": message.ts_ms, "body": message.body,
})).collect::<Vec<_>>(),
"text": payload.text,
});
if let Some(alias) = &payload.room_alias {
out["room_alias"] = json!(alias);
}
Ok(out)
}
fn chat_payload_of(payload: &Value) -> Result<ChatPayload> {
ensure!(payload["corpus"] == CORPUS_CHAT, "a chat read was handed a point of another corpus");
let text = |value: &Value, key: &str| value[key].as_str().map(str::to_owned).with_context(|| format!("a chat chunk carries no {key}"));
let messages = payload["messages"]
.as_array()
.context("a chat chunk carries no messages")?
.iter()
.map(|message| {
Ok(ChatMessage {
event_id: text(message, "event_id")?,
sender: text(message, "sender")?,
ts_ms: message["ts_ms"].as_u64().context("a chat message carries no ts_ms")?,
body: text(message, "body")?,
})
})
.collect::<Result<Vec<_>>>()?;
ensure!(!messages.is_empty(), "a chat chunk carries no messages");
Ok(ChatPayload { room_id: text(payload, "room_id")?, room_alias: payload["room_alias"].as_str().map(str::to_owned), messages, text: text(payload, "text")? })
}
fn payload_json(payload: &SourcePayload) -> Value {
json!({
"corpus": CORPUS_FORGE,
"repo": payload.repo,
"path": payload.path,
"commit": payload.commit,
@ -270,8 +348,17 @@ impl SourceIndex for QdrantIndex {
},
});
self.ask(Method::Put, &format!("{path}?wait=true"), Some(&body), &[200])?;
for field in ["repo", "path"] {
let index = json!({ "field_name": field, "field_schema": "keyword" });
for (field, schema) in [
("repo", "keyword"),
("path", "keyword"),
("corpus", "keyword"),
("room_id", "keyword"),
("room_alias", "keyword"),
("event_ids", "keyword"),
("ts_start_ms", "integer"),
("ts_end_ms", "integer"),
] {
let index = json!({ "field_name": field, "field_schema": schema });
self.ask(Method::Put, &format!("{path}/index?wait=true"), Some(&index), &[200])?;
}
Ok(())
@ -315,9 +402,7 @@ impl SourceIndex for QdrantIndex {
let collection = self.sources.clone();
let http = self.http.bounded(within);
let mut body = json!({ "vector": vector, "limit": limit, "with_payload": true, "with_vector": false });
if let Some(filter) = source_filter(repo, front) {
body["filter"] = filter;
}
body["filter"] = source_filter(repo, front);
let (_, answer) =
self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?;
let found = answer["result"].as_array().context("Qdrant's search has no result")?;
@ -331,6 +416,45 @@ impl SourceIndex for QdrantIndex {
})
.collect()
}
fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> {
let collection = self.sources.clone();
let points: Vec<Value> = points
.iter()
.map(|point| Ok(json!({ "id": point.id, "vector": point.vector, "payload": chat_payload_json(&point.payload)? })))
.collect::<Result<_>>()?;
self.ask(Method::Put, &format!("/collections/{collection}/points?wait=true"), Some(&json!({ "points": points })), &[200])?;
Ok(())
}
fn chat_nearest(&mut self, vector: &[f32], filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatRow>> {
let collection = self.sources.clone();
let http = self.http.bounded(within);
let body = json!({ "vector": vector, "limit": limit, "filter": chat_filter(filter), "with_payload": true, "with_vector": false });
let (_, answer) =
self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/search"), Some(&body), &[200])?;
let found = answer["result"].as_array().context("Qdrant's search has no result")?;
found
.iter()
.map(|point| Ok(ChatRow { score: point["score"].as_f64().context("a hit carries no score")? as f32, payload: chat_payload_of(&point["payload"])? }))
.collect()
}
fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatPayload>> {
let collection = self.sources.clone();
let http = self.http.bounded(within);
let body = json!({
"filter": chat_filter(filter),
"limit": limit,
"order_by": { "key": "ts_start_ms", "direction": "asc" },
"with_payload": true,
"with_vector": false,
});
let (_, answer) =
self.ask_with(&http, Method::Post, &format!("/collections/{collection}/points/scroll"), Some(&body), &[200])?;
let found = answer["result"]["points"].as_array().context("Qdrant's scroll has no points")?;
found.iter().map(|point| chat_payload_of(&point["payload"])).collect()
}
}
#[cfg(test)]
@ -339,9 +463,10 @@ mod source_filter_tests {
/// No ask, no filter; a repo alone is the repo's match.
#[test]
fn no_ask_is_no_filter_and_a_repo_is_its_own_match() {
assert_eq!(source_filter(None, false), None);
assert_eq!(source_filter(Some("Huginn"), false), Some(json!({ "must": [{ "key": "repo", "match": { "value": "Huginn" } }] })));
fn no_ask_is_the_forge_corpus_and_a_repo_is_its_own_match() {
let forge = json!({ "key": "corpus", "match": { "value": "forge" } });
assert_eq!(source_filter(None, false), json!({ "must": [forge] }));
assert_eq!(source_filter(Some("Huginn"), false), json!({ "must": [forge, { "key": "repo", "match": { "value": "Huginn" } }] }));
}
/// The front filter matches the path against exactly the front-page files
@ -354,15 +479,63 @@ mod source_filter_tests {
assert_eq!(path.len(), 1);
path[0]["match"]["any"].as_array().unwrap().iter().map(|p| p.as_str().unwrap().to_owned()).collect()
};
let front = source_filter(None, true).unwrap();
let front = source_filter(None, true);
assert_eq!(any(&front), ["README.md", "README", "readme.md", "README.markdown", "CLAUDE.md", "AGENTS.md"]);
assert_eq!(front["must"].as_array().unwrap().len(), 1, "no repo condition when none was asked");
assert_eq!(front["must"].as_array().unwrap().len(), 2, "the corpus and the path; no repo condition when none was asked");
let both = source_filter(Some("Zyphos"), true).unwrap();
let both = source_filter(Some("Zyphos"), true);
assert_eq!(any(&both), any(&front));
let repos: Vec<&Value> = both["must"].as_array().unwrap().iter().filter(|c| c["key"] == "repo").collect();
assert_eq!(repos.len(), 1);
assert_eq!(repos[0]["match"]["value"], "Zyphos");
assert_eq!(both["must"].as_array().unwrap().len(), 2);
assert_eq!(both["must"].as_array().unwrap().len(), 3);
}
fn payload() -> ChatPayload {
let message = |id: &str, ts: u64| ChatMessage { event_id: id.into(), sender: "@a:gamecult.org".into(), ts_ms: ts, body: format!("body {id}") };
ChatPayload { room_id: "!r:gamecult.org".into(), room_alias: Some("#swarm:gamecult.org".into()), messages: vec![message("$a", 10), message("$b", 20)], text: "swarm\n10 @a: body".into() }
}
/// The chat codec is its own inverse, writes the contract's fields derived
/// from the messages, and refuses a chunk with none and a point of another corpus.
#[test]
fn the_chat_payload_round_trips_the_contract() {
let written = chat_payload_json(&payload()).unwrap();
assert_eq!(written["corpus"], "chat");
assert_eq!((written["ts_start_ms"].as_u64(), written["ts_end_ms"].as_u64()), (Some(10), Some(20)));
assert_eq!((written["first_event_id"].as_str(), written["last_event_id"].as_str()), (Some("$a"), Some("$b")));
assert_eq!(written["event_ids"], json!(["$a", "$b"]));
assert_eq!(written["room_alias"], "#swarm:gamecult.org");
assert_eq!(chat_payload_of(&written).unwrap(), payload());
let mut no_alias = payload();
no_alias.room_alias = None;
assert!(chat_payload_json(&no_alias).unwrap().get("room_alias").is_none());
assert_eq!(chat_payload_of(&chat_payload_json(&no_alias).unwrap()).unwrap(), no_alias);
assert!(chat_payload_json(&ChatPayload { messages: vec![], ..payload() }).is_err());
let forge = payload_json(&SourcePayload { repo: "r".into(), path: "p".into(), commit: "c".into(), blob: "b".into(), line_start: 1, line_end: 2, chunk_index: 0, ext: "rs".into(), text: "t".into() });
assert_eq!(forge["corpus"], "forge");
assert!(chat_payload_of(&forge).is_err());
}
/// The chat filter is the chat corpus, the rooms as an any-of `should` split
/// by id and alias, and the two ranges that keep an overlapping span.
#[test]
fn the_chat_filter_is_corpus_rooms_and_overlap() {
let filter = chat_filter(&ChatFilter { rooms: vec!["!r:x".into(), "#a:x".into(), "#b:x".into()], after_ms: Some(5), before_ms: Some(9) });
assert_eq!(
filter,
json!({
"must": [
{ "key": "corpus", "match": { "value": "chat" } },
{ "key": "ts_end_ms", "range": { "gte": 5 } },
{ "key": "ts_start_ms", "range": { "lte": 9 } },
],
"should": [
{ "key": "room_id", "match": { "any": ["!r:x"] } },
{ "key": "room_alias", "match": { "any": ["#a:x", "#b:x"] } },
],
})
);
assert_eq!(chat_filter(&ChatFilter::default()), json!({ "must": [{ "key": "corpus", "match": { "value": "chat" } }] }));
}
}

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

File diff suppressed because it is too large Load diff

190
crates/huginn-daemon/src/sources/mod.rs
View file

@ -1,5 +1,7 @@
//! The source index: the public GameCult forge repos' code and docs as chunks
//! in one Qdrant collection, `huginn_sources`, for semantic search.
//! The source index: the public GameCult forge repos' code and docs, and the
//! Minds' chat rooms, as chunks in one Qdrant collection, `huginn_sources`,
//! for semantic search. Each point's `corpus` payload (`forge` or `chat`) keeps
//! the two apart; a forge search filters `forge` and a chat search `chat`.
//!
//! Two truths, one writer. The forge is the truth of which repos exist and what
//! they hold; the collection is derived from it and can be deleted whenever.
@ -30,7 +32,7 @@ use std::path::Path;
use std::time::{Duration, Instant};
use anyhow::{Context, Result, bail, ensure};
use huginn_mind::{SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery};
use huginn_mind::{ChatHit, ChatHits, ChatMessage, ChatQuery, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery};
use crate::index::{BATCH, BULK_EMBED_TIMEOUT, Budget, Embedder, INDEX_CALL_TIMEOUT, MANAGED_BY, ModelIdentity, point_id};
use chunk::{CHUNKER_VERSION, embedded_text};
@ -43,8 +45,19 @@ pub const COLLECTION: &str = "huginn_sources";
/// agent reads first to learn what the repo is. The only owner of the list.
pub const FRONT_PAGE_FILES: [&str; 6] = ["README.md", "README", "readme.md", "README.markdown", "CLAUDE.md", "AGENTS.md"];
/// What the collection says its corpus is.
pub const CORPUS: &str = "forge:GameCult";
/// What the collection says its corpus is: the set of corpora it holds, each
/// point's `corpus` payload naming which (`forge` or `chat`). A collection
/// labelled with any other set is rebuilt.
pub const CORPUS: &str = "forge:GameCult+chat:rooms";
/// The `corpus` payload of a forge point, and of a chat point.
pub const CORPUS_FORGE: &str = "forge";
pub const CORPUS_CHAT: &str = "chat";
/// The instruction Qwen3 embedding models take in front of a question about
/// past conversation.
pub const CHAT_QUERY_INSTRUCTION: &str =
"Given a question about past conversation in chat rooms, retrieve the chat passages that answer it";
/// The instruction Qwen3 embedding models take in front of a query about code.
pub const SOURCE_QUERY_INSTRUCTION: &str =
@ -111,6 +124,58 @@ pub struct SourcePoint {
pub payload: SourcePayload,
}
/// What travels beside a chat chunk's vector: consecutive messages of one
/// room, never spanning rooms. The single owner of the chat point's shape; the
/// ingest job is its only writer (through `upsert_chat`) and nothing else of the
/// payload is read. `messages` are the display truth and hold, in order, at
/// least one message. The chunk's span, its first and last event ids and its
/// `event_ids` are derived from them by the codec, so they cannot disagree.
/// `text` is what was embedded: the room's name, then `time sender: body` per
/// message.
#[derive(Clone, Debug, PartialEq)]
pub struct ChatPayload {
pub room_id: String,
/// Absent when the room has no canonical alias.
pub room_alias: Option<String>,
pub messages: Vec<ChatMessage>,
pub text: String,
}
#[derive(Clone, Debug, PartialEq)]
pub struct ChatPoint {
pub id: String,
pub vector: Vec<f32>,
pub payload: ChatPayload,
}
/// One chat chunk a vector search returned and how near it scored.
#[derive(Clone, Debug, PartialEq)]
pub struct ChatRow {
pub score: f32,
pub payload: ChatPayload,
}
/// Which chat chunks a read keeps; the selectors compose as AND. `rooms` are
/// any-of room ids (`!`) or aliases (`#`); the span keeps chunks whose own
/// span overlaps [`after_ms`, `before_ms`].
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ChatFilter {
pub rooms: Vec<String>,
pub after_ms: Option<u64>,
pub before_ms: Option<u64>,
}
impl ChatFilter {
/// Whether a chunk of `room_id`/`room_alias` spanning [`start`, `end`]
/// passes. The fake index and the tests read the filter through this; the
/// Qdrant index states the same thing as conditions.
pub fn keeps(&self, room_id: &str, room_alias: Option<&str>, start: u64, end: u64) -> bool {
(self.rooms.is_empty() || self.rooms.iter().any(|room| room == room_id || Some(room.as_str()) == room_alias))
&& self.after_ms.is_none_or(|after| end >= after)
&& self.before_ms.is_none_or(|before| start <= before)
}
}
/// One chunk a vector search returned and how near it scored.
#[derive(Clone, Debug, PartialEq)]
pub struct SourceRow {
@ -124,7 +189,9 @@ pub struct SourceRow {
pub trait SourceIndex {
fn describe_sources(&mut self, within: Duration) -> Result<DescribedSources>;
/// Drops the collection if it exists and creates it empty with this
/// metadata, no cursors, and keyword indexes on `repo` and `path`.
/// metadata, no cursors, keyword indexes on `repo`, `path`, `corpus`,
/// `room_id`, `room_alias` and `event_ids`, and integer indexes on
/// `ts_start_ms` and `ts_end_ms`.
fn recreate_sources(&mut self, meta: &SourcesMeta) -> Result<()>;
/// Replaces the cursors object whole.
fn set_cursors(&mut self, cursors: &BTreeMap<String, String>) -> Result<()>;
@ -137,6 +204,15 @@ pub trait SourceIndex {
/// The `limit` nearest chunks, best first, in `repo` alone when given, and
/// among the `FRONT_PAGE_FILES` alone when `front`.
fn search_sources(&mut self, vector: &[f32], repo: Option<&str>, front: bool, limit: u32, within: Duration) -> Result<Vec<SourceRow>>;
/// Writes chat chunks, acknowledged before it returns. The ingest job's
/// alone: the daemon never calls it. A chunk with no messages is refused.
fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()>;
/// The `limit` chat chunks nearest `vector`, best first, among those
/// `filter` keeps. Reads only `corpus=chat` points; the daemon's.
fn chat_nearest(&mut self, vector: &[f32], filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatRow>>;
/// Up to `limit` chat chunks `filter` keeps, oldest first by span start.
/// Reads only `corpus=chat` points; the daemon's.
fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatPayload>>;
}
/// A repo the forge lists as indexable.
@ -476,6 +552,10 @@ pub fn fuse<T>(lists: Vec<Vec<T>>, key: impl Fn(&T) -> String, k: u32, top_k: us
/// of one, trimmed, and each distinct text once, in the order asked.
fn query_texts(query: &SourceQuery) -> Vec<&str> {
let asked: Vec<&str> = if query.texts.is_empty() { vec![query.text.0.as_str()] } else { query.texts.iter().map(|text| text.0.as_str()).collect() };
distinct_texts(asked)
}
fn distinct_texts(asked: Vec<&str>) -> Vec<&str> {
let mut distinct: Vec<&str> = Vec::with_capacity(asked.len());
for text in asked {
if !distinct.iter().any(|seen| seen.trim() == text.trim()) {
@ -485,16 +565,12 @@ fn query_texts(query: &SourceQuery) -> Vec<&str> {
distinct
}
/// A source search, which a daemon answers from the collection as it finds it:
/// it embeds the query with its own embedder, and refuses if the collection
/// was built under another model, because vectors of two models do not compare.
pub fn search<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, query: &SourceQuery, within: Duration) -> Result<SourceHits> {
let budget = Budget::new(within);
let built = match index.describe_sources(budget.left()?).context("describing the source collection")? {
DescribedSources::Absent => bail!("source index not built"),
DescribedSources::Unlabelled => bail!("source index not built: {COLLECTION} carries no Huginn metadata"),
DescribedSources::Labelled { meta, .. } => meta,
};
/// The label of the collection a daemon is about to read, if it was built
/// under the model this daemon embeds with. Absent, unlabelled and other-model
/// collections are loud refusals: vectors of two models do not compare, and an
/// empty page is not an answer.
fn readable_under_this_model<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, budget: &Budget) -> Result<(SourcesMeta, ModelIdentity)> {
let built = built_label(index, budget)?;
let identity = embedder.model_identity(budget.left()?).context("reading the embedding model's identity")?;
ensure!(
built.model == identity.name && built.model_digest == identity.digest && built.dimensions == identity.dimensions,
@ -506,6 +582,23 @@ pub fn search<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, quer
identity.digest,
identity.dimensions
);
Ok((built, identity))
}
fn built_label<S: SourceIndex>(index: &mut S, budget: &Budget) -> Result<SourcesMeta> {
match index.describe_sources(budget.left()?).context("describing the source collection")? {
DescribedSources::Absent => bail!("source index not built"),
DescribedSources::Unlabelled => bail!("source index not built: {COLLECTION} carries no Huginn metadata"),
DescribedSources::Labelled { meta, .. } => Ok(meta),
}
}
/// A source search, which a daemon answers from the collection as it finds it:
/// it embeds the query with its own embedder, and refuses if the collection
/// was built under another model, because vectors of two models do not compare.
pub fn search<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, query: &SourceQuery, within: Duration) -> Result<SourceHits> {
let budget = Budget::new(within);
let (built, identity) = readable_under_this_model(embedder, index, &budget)?;
let topics = query_texts(query);
let prompts: Vec<String> = topics.iter().map(|topic| format!("Instruct: {SOURCE_QUERY_INSTRUCTION}\nQuery: {topic}")).collect();
let vectors = embedder.embed(&prompts, budget.left()?).context("embedding the query")?;
@ -539,3 +632,68 @@ pub fn search<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, quer
.collect(),
})
}
/// A chat search. With texts, the chunks nearest them among those the room and
/// span selectors keep, several texts fused by `fuse`; with none, a listing of
/// what the selectors keep, oldest first, `top_k` of it and `truncated` saying
/// more exist (no embedding, so the model's identity is not consulted, though
/// an absent collection is still refused). A hit outside the span is not
/// returned and a boundary hit loses its messages outside [after, before], its
/// span and first and last ids recomputed from what is left. Never writes.
pub fn search_chat<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, query: &ChatQuery, within: Duration) -> Result<ChatHits> {
let budget = Budget::new(within);
let filter = ChatFilter { rooms: query.rooms.iter().map(|room| room.0.clone()).collect(), after_ms: query.after_ms, before_ms: query.before_ms };
let top_k = query.top_k as usize;
let topics = distinct_texts(query.texts.iter().map(|text| text.0.as_str()).collect());
let (built, rows, truncated) = if topics.is_empty() {
let built = built_label(index, &budget)?;
let mut held = index.chat_scroll(&filter, query.top_k + 1, budget.left()?)?;
let truncated = held.len() > top_k;
held.truncate(top_k);
(built, held.into_iter().map(|payload| (None, payload)).collect::<Vec<_>>(), truncated)
} else {
let (built, identity) = readable_under_this_model(embedder, index, &budget)?;
let prompts: Vec<String> = topics.iter().map(|topic| format!("Instruct: {CHAT_QUERY_INSTRUCTION}\nQuery: {topic}")).collect();
let vectors = embedder.embed(&prompts, budget.left()?).context("embedding the query")?;
ensure!(vectors.len() == prompts.len(), "the embedder returned {} vectors for {} query texts", vectors.len(), prompts.len());
let depth = if vectors.len() == 1 { query.top_k } else { FUSE_DEPTH };
let mut lists = Vec::with_capacity(vectors.len());
for vector in &vectors {
ensure!(vector.len() == identity.dimensions as usize, "the embedder returned a query vector of {} where the collection holds {}", vector.len(), identity.dimensions);
lists.push(index.chat_nearest(vector, &filter, depth, budget.left()?)?);
}
let rows: Vec<ChatRow> = if lists.len() == 1 {
lists.remove(0)
} else {
let key = |row: &ChatRow| format!("{}\0{}", row.payload.room_id, row.payload.messages.first().map_or("", |message| message.event_id.as_str()));
fuse(lists, key, FUSE_K, top_k).into_iter().map(|(score, row)| ChatRow { score, ..row }).collect()
};
(built, rows.into_iter().map(|row| (Some(row.score), row.payload)).collect(), false)
};
Ok(ChatHits {
indexed_under: SourceModel { name: built.model, digest: built.model_digest },
truncated,
hits: rows.into_iter().filter_map(|(score, payload)| chat_hit(score, payload, &filter)).collect(),
})
}
/// A held chunk as a hit: only the messages inside the asked span, the span and
/// ids those messages make, and nothing at all when none is inside.
fn chat_hit(score: Option<f32>, payload: ChatPayload, filter: &ChatFilter) -> Option<ChatHit> {
let messages: Vec<ChatMessage> = payload
.messages
.into_iter()
.filter(|message| filter.after_ms.is_none_or(|after| message.ts_ms >= after) && filter.before_ms.is_none_or(|before| message.ts_ms <= before))
.collect();
let (first, last) = (messages.first()?, messages.last()?);
Some(ChatHit {
room_id: payload.room_id,
room_alias: payload.room_alias,
first_event_id: first.event_id.clone(),
last_event_id: last.event_id.clone(),
ts_start_ms: messages.iter().map(|message| message.ts_ms).min()?,
ts_end_ms: messages.iter().map(|message| message.ts_ms).max()?,
score,
messages,
})
}

183
crates/huginn-daemon/src/sources/tests.rs
View file

@ -7,7 +7,7 @@ use std::process::Command;
use anyhow::Result;
use huginn_mind::eureka_pipeline::{Line, Short};
use huginn_mind::SourceQuery;
use huginn_mind::{ChatHits, ChatQuery, SourceQuery};
use super::*;
use crate::index::fakes::{FakeEmbedder, FakeSources};
@ -755,3 +755,184 @@ fn a_long_brief_of_eight_topics_reaches_the_chunk_only_the_last_is_near() {
assert!(paths(&found).contains(&"chunk-7.rs"), "{:?}", paths(&found));
assert_eq!(paths(&found).len(), 8);
}
// ---- chat search ----
fn msg(event_id: &str, ts_ms: u64) -> ChatMessage {
ChatMessage { event_id: event_id.into(), sender: "@a:gamecult.org".into(), ts_ms, body: format!("body of {event_id}") }
}
/// A chunk in `room` (with `alias`) of messages `(id, ts)`, near `axis`.
fn chat_point(room: &str, alias: Option<&str>, axis: usize, dimensions: u32, messages: &[(&str, u64)]) -> ChatPoint {
let messages: Vec<ChatMessage> = messages.iter().map(|(id, ts)| msg(id, *ts)).collect();
ChatPoint {
id: point_id(&format!("{room}{}", messages[0].event_id)),
vector: (0..dimensions as usize).map(|at| if at == axis { 1.0 } else { 0.0 }).collect(),
payload: ChatPayload { room_id: room.into(), room_alias: alias.map(str::to_owned), messages, text: format!("chunk near topic-{axis}") },
}
}
fn chat_query(texts: &[&str], rooms: &[&str], after: Option<u64>, before: Option<u64>, top_k: u32) -> ChatQuery {
ChatQuery {
texts: texts.iter().map(|text| Line((*text).into())).collect(),
rooms: rooms.iter().map(|room| Short((*room).into())).collect(),
after_ms: after,
before_ms: before,
top_k,
}
}
/// A labelled collection that also holds the chat chunks given. Two rooms:
/// `!a` (alias `#swarm`), `!b` (no alias).
fn chat_rig(points: Vec<ChatPoint>) -> (Rig, TopicEmbedder) {
let (mut rig, embedder) = topic_rig(4);
rig.index.state.lock().unwrap().points.clear();
rig.index.upsert_chat(&points).unwrap();
(rig, embedder)
}
fn ids(hits: &ChatHits) -> Vec<&str> {
hits.hits.iter().map(|hit| hit.first_event_id.as_str()).collect()
}
const WITHIN: Duration = Duration::from_secs(5);
fn three_rooms() -> Vec<ChatPoint> {
vec![
chat_point("!a:x", Some("#swarm:x"), 0, 4, &[("$a1", 100), ("$a2", 200)]),
chat_point("!a:x", Some("#swarm:x"), 1, 4, &[("$a3", 300), ("$a4", 400)]),
chat_point("!b:x", None, 0, 4, &[("$b1", 150)]),
]
}
#[test]
fn a_room_filter_by_id_and_by_alias_returns_only_that_room() {
let (mut rig, mut embedder) = chat_rig(three_rooms());
let by_alias = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x"], None, None, 10), WITHIN).unwrap();
assert_eq!(ids(&by_alias), ["$a1", "$a3"]);
let by_id = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["!b:x"], None, None, 10), WITHIN).unwrap();
assert_eq!(ids(&by_id), ["$b1"], "a room with no alias is reachable by id");
assert_eq!(by_id.hits[0].room_alias, None);
let both = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x", "!b:x"], None, None, 10), WITHIN).unwrap();
assert_eq!(ids(&both), ["$a1", "$b1", "$a3"], "any-of, oldest first");
let nobody = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#nobody:x"], None, None, 10), WITHIN).unwrap();
assert!(nobody.hits.is_empty() && !nobody.truncated, "an unknown alias is zero hits, not an error");
}
#[test]
fn a_time_span_returns_the_overlapping_chunks_and_trims_boundary_messages() {
let (mut rig, mut embedder) = chat_rig(three_rooms());
let hits = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x"], Some(150), Some(350), 10), WITHIN).unwrap();
assert_eq!(ids(&hits), ["$a2", "$a3"], "the boundary chunks keep only in-span messages and first ids are recomputed");
let (first, second) = (&hits.hits[0], &hits.hits[1]);
assert_eq!((first.ts_start_ms, first.ts_end_ms, first.last_event_id.as_str(), first.messages.len()), (200, 200, "$a2", 1));
assert_eq!((second.ts_start_ms, second.ts_end_ms, second.last_event_id.as_str(), second.messages.len()), (300, 300, "$a3", 1));
let none = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["#swarm:x"], Some(1000), None, 10), WITHIN).unwrap();
assert!(none.hits.is_empty(), "a chunk wholly outside is absent");
}
#[test]
fn list_mode_is_oldest_first_truncated_when_more_exist_and_needs_no_embed() {
let (mut rig, mut embedder) = chat_rig(three_rooms());
let page = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &[], None, None, 2), WITHIN).unwrap();
assert_eq!(ids(&page), ["$a1", "$b1"]);
assert!(page.truncated);
assert!(embedder.0.state.lock().unwrap().embedded.is_empty(), "no texts, no embed call");
// Within one room chunks never overlap, so the last hit's end + 1 continues without a repeat.
let first = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["!a:x"], None, None, 1), WITHIN).unwrap();
assert_eq!((ids(&first), first.truncated), (vec!["$a1"], true));
let rest = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &["!a:x"], Some(first.hits[0].ts_end_ms + 1), None, 1), WITHIN).unwrap();
assert_eq!((ids(&rest), rest.truncated), (vec!["$a3"], false));
}
#[test]
fn texts_rank_by_meaning_and_compose_with_room_and_span() {
let (mut rig, mut embedder) = chat_rig(three_rooms());
let near = search_chat(&mut embedder, &mut rig.index, &chat_query(&["topic-0"], &[], None, None, 5), WITHIN).unwrap();
assert_eq!(near.hits.len(), 3);
assert!(near.hits[0].score.unwrap() > near.hits[2].score.unwrap());
assert_eq!(near.hits[2].first_event_id, "$a3", "the far chunk is last");
let narrowed = search_chat(&mut embedder, &mut rig.index, &chat_query(&["topic-0"], &["!a:x"], Some(250), None, 5), WITHIN).unwrap();
assert_eq!(ids(&narrowed), ["$a3"], "a near chunk in another room or outside the span never appears");
let fused = search_chat(&mut embedder, &mut rig.index, &chat_query(&["topic-0", "topic-1"], &["!a:x"], None, None, 5), WITHIN).unwrap();
assert_eq!(fused.hits.len(), 2);
assert!(fused.hits[0].score.unwrap() > 0.0 && fused.hits[0].score != Some(1.0), "two texts carry the fused reciprocal-rank score");
assert_eq!(embedder.0.state.lock().unwrap().embedded.last().unwrap().len(), 2, "one embed call for two texts");
}
#[test]
fn every_hit_cites_room_event_ids_and_timestamps() {
let (mut rig, mut embedder) = chat_rig(three_rooms());
for texts in [&[][..], &["topic-0"][..]] {
let hits = search_chat(&mut embedder, &mut rig.index, &chat_query(texts, &[], None, None, 10), WITHIN).unwrap();
assert!(!hits.hits.is_empty());
for hit in &hits.hits {
assert!(hit.room_id.starts_with('!') && !hit.first_event_id.is_empty() && !hit.last_event_id.is_empty());
assert!(hit.ts_start_ms <= hit.ts_end_ms && !hit.messages.is_empty());
assert_eq!((hit.messages[0].event_id.as_str(), hit.messages.last().unwrap().event_id.as_str()), (hit.first_event_id.as_str(), hit.last_event_id.as_str()));
for message in &hit.messages {
assert!(message.event_id.starts_with('$') && message.sender.starts_with('@') && message.ts_ms > 0);
}
}
}
}
#[test]
fn a_forge_search_never_returns_a_chat_point_and_a_chat_search_never_a_forge_point() {
let (mut rig, mut embedder) = topic_rig(4);
rig.index.upsert_chat(&three_rooms()).unwrap();
let forge = search(&mut embedder, &mut rig.index, &queries(&["topic-0"], 10), WITHIN).unwrap();
assert_eq!(forge.hits.len(), 4, "only the four forge chunks");
let chat = search_chat(&mut embedder, &mut rig.index, &chat_query(&["topic-0"], &[], None, None, 10), WITHIN).unwrap();
assert_eq!(chat.hits.len(), 3, "only the three chat chunks");
let listed = search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &[], None, None, 10), WITHIN).unwrap();
assert_eq!(listed.hits.len(), 3);
}
#[test]
fn an_absent_unlabelled_or_other_model_collection_is_a_loud_refusal_for_chat() {
let mut rig = Rig::new();
let mut embedder = TopicEmbedder(FakeEmbedder::new("d1"));
for texts in [&[][..], &["topic-0"][..]] {
let error = search_chat(&mut embedder, &mut rig.index, &chat_query(texts, &[], None, None, 3), WITHIN).unwrap_err();
assert_eq!(format!("{error:#}"), "source index not built");
}
rig.index.plant(DescribedSources::Unlabelled);
assert!(search_chat(&mut embedder, &mut rig.index, &chat_query(&[], &[], None, None, 3), WITHIN).is_err());
let (mut rig, _) = chat_rig(three_rooms());
let mut foreign = TopicEmbedder(FakeEmbedder::new("d2"));
let error = format!("{:#}", search_chat(&mut foreign, &mut rig.index, &chat_query(&["topic-0"], &[], None, None, 3), WITHIN).unwrap_err());
assert!(error.contains("built under model d1") && error.contains("embeds under model d2"), "{error}");
}
/// An old forge-only collection (labelled with the old corpus) is rebuilt once
/// under the new corpus, and a second refresh leaves it.
#[test]
fn an_old_forge_only_collection_rebuilds_once() {
let remote = Remote::new();
remote.commit(&[("a.rs", b"one\n")], &[]);
let mut rig = Rig::new();
let mut forge = FakeForge::of(&[("Demo", &remote)]);
rig.converged(&mut forge);
let mut old = match rig.index.state.lock().unwrap().described.clone() {
Some(DescribedSources::Labelled { meta, cursors }) => (meta, cursors),
other => panic!("{other:?}"),
};
assert_eq!(old.0.corpus, "forge:GameCult+chat:rooms");
old.0.corpus = "forge:GameCult".into();
rig.index.plant(DescribedSources::Labelled { meta: old.0, cursors: old.1 });
let before = rig.index.state.lock().unwrap().recreated;
rig.converged(&mut forge);
rig.converged(&mut forge);
assert_eq!(rig.index.state.lock().unwrap().recreated, before + 1);
}
#[test]
fn the_chat_read_path_holds_no_discord_matrix_or_second_fuse() {
let source = include_str!("mod.rs");
let read = &source[source.find("pub fn search_chat").unwrap()..];
for banned in ["discord", "puppet", "bridge", "matrix", "synapse", "access_token"] {
assert!(!read.to_ascii_lowercase().contains(banned), "{banned} in the chat read path");
}
assert_eq!(source.matches("pub fn fuse").count(), 1);
}

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

@ -72,7 +72,7 @@ pub use store::{MindStore, OwnedRedbMessagePackBackingStore};
pub use wire::{
DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MAX_DEFERRED_BODY_BYTES, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON,
MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus, PersonaStatus,
ACTIVITY_REPOS_MAX, OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent, RepoMoved, SOURCE_TEXTS_MAX, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery,
ACTIVITY_REPOS_MAX, CHAT_ROOMS_MAX, ChatHit, ChatHits, ChatMessage, ChatQuery, OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent, RepoMoved, SOURCE_TEXTS_MAX, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery,
};
/// The leaf, whole, through the one crate that pins its rev: the daemon and

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

@ -92,6 +92,10 @@ pub enum HuginnMindRequest {
/// docs, which no mind holds. The daemon embeds the text and asks the
/// index it does not write.
SearchSources { instance: Slug, query: SourceQuery },
/// Search of the chat index: chunks of the rooms the Minds are meant to
/// have, by room, by time span and by meaning. Read-only, like
/// `SearchSources`, and queued with it.
SearchChat { instance: Slug, query: ChatQuery },
/// What others did in `repos` on the forge since the `after` cursor: the
/// daemon reads the forge and no document of the mind, and answers
/// `Refused(Unavailable)` when it has no forge or the forge does not answer
@ -270,6 +274,96 @@ pub struct SourceHits {
pub hits: Vec<SourceHit>,
}
/// The most rooms one chat search names.
pub const CHAT_ROOMS_MAX: usize = 16;
/// A chat search. The selectors compose as AND. `rooms` are any-of: a value
/// beginning `!` is a room id, `#` a canonical alias. `after_ms` and
/// `before_ms` (Unix milliseconds) keep the chunks whose span overlaps
/// [after, before]. `texts` (none to `SOURCE_TEXTS_MAX`) is the meaning match,
/// several fused as in a source search; with none the search is a listing:
/// chunks oldest first, at most `top_k`, `ChatHits.truncated` saying more
/// exist, and the caller narrows with `after_ms` = the last hit's
/// `ts_end_ms + 1`.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ChatQuery {
pub texts: Vec<Line>,
pub rooms: Vec<Short>,
pub after_ms: Option<u64>,
pub before_ms: Option<u64>,
pub top_k: u32,
}
impl ChatQuery {
/// The door. A refusal names the field.
pub fn check(&self) -> Result<(), MindRefusal> {
let refused = |field: &str, value: Option<String>, message: &str| MindRefusal::SelectionInvalid {
field: field.into(),
value,
message: message.into(),
};
if self.texts.len() > SOURCE_TEXTS_MAX {
return Err(refused("texts", Some(self.texts.len().to_string()), "a chat search takes 0 to 8 texts"));
}
if self.texts.iter().any(|text| text.0.trim().is_empty()) {
return Err(refused("texts", None, "every text of a chat search says something"));
}
if self.rooms.len() > CHAT_ROOMS_MAX {
return Err(refused("rooms", Some(self.rooms.len().to_string()), "a chat search names at most 16 rooms"));
}
if let Some(room) = self.rooms.iter().find(|room| !(room.0.starts_with('!') || room.0.starts_with('#')) || room.0.len() < 2) {
return Err(refused("rooms", Some(room.0.clone()), "a room is a room id beginning `!` or a canonical alias beginning `#`"));
}
if let (Some(after), Some(before)) = (self.after_ms, self.before_ms)
&& before < after
{
return Err(refused("before_ms", Some(before.to_string()), "before_ms is not earlier than after_ms"));
}
if !(1..=SOURCE_TOP_K_MAX).contains(&self.top_k) {
return Err(refused("top_k", Some(self.top_k.to_string()), "a chat search asks for 1 to 50 hits"));
}
Ok(())
}
}
/// One message of a chunk, as the room held it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ChatMessage {
pub event_id: String,
/// The sender's Matrix id.
pub sender: String,
pub ts_ms: u64,
pub body: String,
}
/// One chunk of consecutive messages in one room. Every citation is
/// required: only the room's alias may be absent, because a room may have
/// none. The span and the first and last ids are those of `messages` as
/// returned, after any trimming to the asked span.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct ChatHit {
pub room_id: String,
pub room_alias: Option<String>,
pub first_event_id: String,
pub last_event_id: String,
pub ts_start_ms: u64,
pub ts_end_ms: u64,
/// Cosine similarity for one text, the reciprocal-rank sum for several,
/// none in a listing.
pub score: Option<f32>,
pub messages: Vec<ChatMessage>,
}
/// The answer to a chat search: oldest first in a listing, nearest first with
/// texts.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct ChatHits {
pub indexed_under: SourceModel,
/// A listing had more chunks than `top_k`.
pub truncated: bool,
pub hits: Vec<ChatHit>,
}
impl HuginnMindRequest {
/// The envelope's `operation`.
pub fn operation(&self) -> &'static str {
@ -281,6 +375,7 @@ impl HuginnMindRequest {
Self::PersonaGet { .. } => "persona_get",
Self::PersonaPut { .. } => "persona_put",
Self::SearchSources { .. } => "search_sources",
Self::SearchChat { .. } => "search_chat",
Self::StewardedActivity { .. } => "stewarded_activity",
Self::RootsTouched { .. } => "roots_touched",
}
@ -296,6 +391,7 @@ impl HuginnMindRequest {
| Self::PersonaGet { instance }
| Self::PersonaPut { instance, .. }
| Self::SearchSources { instance, .. }
| Self::SearchChat { instance, .. }
| Self::StewardedActivity { instance, .. }
| Self::RootsTouched { instance } => Some(instance),
}
@ -315,6 +411,7 @@ pub enum HuginnMindResponse {
Persona(Option<PersonaStateView>),
PersonaPut(PersonaPutOutcome),
Sources(SourceHits),
Chat(ChatHits),
Activity(RepoActivity),
RootsTouched(Vec<RootTouched>),
/// The answer did not fit one send and is being held for the client to
@ -450,6 +547,10 @@ mod tests {
instance: slug(INSTANCE),
query: SourceQuery { text: Line::default(), texts: vec![Line("reconcile".into()), Line("fuse".into())], top_k: 5, repo: None, front: false },
},
HuginnMindRequest::SearchChat {
instance: slug(INSTANCE),
query: ChatQuery { texts: vec![Line("release".into())], rooms: vec![Short("#swarm:gamecult.org".into())], after_ms: Some(1), before_ms: None, top_k: 5 },
},
HuginnMindRequest::RootsTouched { instance: slug(INSTANCE) },
]
}
@ -500,6 +601,20 @@ mod tests {
text: "x".into(),
}],
}),
HuginnMindResponse::Chat(ChatHits {
indexed_under: SourceModel { name: "m".into(), digest: "d".into() },
truncated: false,
hits: vec![ChatHit {
room_id: "!r:gamecult.org".into(),
room_alias: None,
first_event_id: "$a".into(),
last_event_id: "$a".into(),
ts_start_ms: 1,
ts_end_ms: 1,
score: None,
messages: vec![ChatMessage { event_id: "$a".into(), sender: "@x:gamecult.org".into(), ts_ms: 1, body: "hi".into() }],
}],
}),
HuginnMindResponse::RootsTouched(seeded().roots_touched().unwrap()),
HuginnMindResponse::Deferred(DeferredAnswer {
manifest: cultnet_rs::pack_content("huginn.mind_response", "package", "", "", "", b"body", 4)
@ -515,7 +630,7 @@ mod tests {
#[test]
fn the_wire_vocabulary_is_the_minds_methods_and_status_is_derived_from_the_response() {
let names: Vec<&str> = requests().iter().map(|request| request.operation()).collect();
assert_eq!(names, ["whoami", "admit", "view", "query", "persona_get", "persona_put", "search_sources", "search_sources", "roots_touched"]);
assert_eq!(names, ["whoami", "admit", "view", "query", "persona_get", "persona_put", "search_sources", "search_sources", "search_chat", "roots_touched"]);
for request in requests() {
let declared = request.instance().is_none();
assert_eq!(declared, matches!(request, HuginnMindRequest::Whoami), "{request:?}");
@ -625,6 +740,33 @@ mod tests {
assert_eq!((page.matched, page.as_of, page.scores), (0, 7, None));
}
/// A chat search is refused by field at the door, and the published
/// response marks every citation required.
#[test]
fn a_chat_query_is_refused_by_field_and_a_hit_cites_room_ids_and_times() {
let ok = ChatQuery { texts: vec![], rooms: vec![Short("!a:b".into()), Short("#c:b".into())], after_ms: Some(1), before_ms: Some(1), top_k: 50 };
assert!(ok.check().is_ok());
let field = |query: ChatQuery| match query.check() {
Err(MindRefusal::SelectionInvalid { field, .. }) => field,
other => panic!("{other:?}"),
};
assert_eq!(field(ChatQuery { rooms: vec![Short("swarm".into())], ..ok.clone() }), "rooms");
assert_eq!(field(ChatQuery { rooms: vec![Short("#".into())], ..ok.clone() }), "rooms");
assert_eq!(field(ChatQuery { before_ms: Some(0), ..ok.clone() }), "before_ms");
assert_eq!(field(ChatQuery { top_k: 0, ..ok.clone() }), "top_k");
assert_eq!(field(ChatQuery { top_k: 51, ..ok.clone() }), "top_k");
assert_eq!(field(ChatQuery { texts: vec![Line(" ".into())], ..ok.clone() }), "texts");
assert_eq!(field(ChatQuery { texts: vec![Line("x".into()); 9], ..ok.clone() }), "texts");
assert_eq!(field(ChatQuery { rooms: vec![Short("!a:b".into()); 17], ..ok }), "rooms");
let response = serde_json::to_value(schemars::schema_for!(HuginnMindResponse)).unwrap();
let required: Vec<&str> = response["$defs"]["ChatHit"]["required"].as_array().unwrap().iter().filter_map(|v| v.as_str()).collect();
for field in ["room_id", "first_event_id", "last_event_id", "ts_start_ms", "ts_end_ms", "messages"] {
assert!(required.contains(&field), "{field} must be required");
}
let message: Vec<&str> = response["$defs"]["ChatMessage"]["required"].as_array().unwrap().iter().filter_map(|v| v.as_str()).collect();
assert_eq!(message, ["event_id", "sender", "ts_ms", "body"]);
}
/// The two files the catalog publishes are the derivation of the two
/// types, byte for byte.
#[test]

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

@ -158,6 +158,31 @@
"SearchSources"
]
},
{
"description": "Search of the chat index: chunks of the rooms the Minds are meant to\nhave, by room, by time span and by meaning. Read-only, like\n`SearchSources`, and queued with it.",
"type": "object",
"properties": {
"SearchChat": {
"type": "object",
"properties": {
"instance": {
"$ref": "#/$defs/Slug"
},
"query": {
"$ref": "#/$defs/ChatQuery"
}
},
"required": [
"instance",
"query"
]
}
},
"additionalProperties": false,
"required": [
"SearchChat"
]
},
{
"description": "What others did in `repos` on the forge since the `after` cursor: the\ndaemon reads the forge and no document of the mind, and answers\n`Refused(Unavailable)` when it has no forge or the forge does not answer\nin time. The repos are the asker's to name (her stewardships); the\ninstance's own login is dropped from the answer.",
"type": "object",
@ -272,6 +297,50 @@
"deletion_line"
]
},
"ChatQuery": {
"description": "A chat search. The selectors compose as AND. `rooms` are any-of: a value\nbeginning `!` is a room id, `#` a canonical alias. `after_ms` and\n`before_ms` (Unix milliseconds) keep the chunks whose span overlaps\n[after, before]. `texts` (none to `SOURCE_TEXTS_MAX`) is the meaning match,\nseveral fused as in a source search; with none the search is a listing:\nchunks oldest first, at most `top_k`, `ChatHits.truncated` saying more\nexist, and the caller narrows with `after_ms` = the last hit's\n`ts_end_ms + 1`.",
"type": "object",
"properties": {
"after_ms": {
"type": [
"integer",
"null"
],
"format": "uint64",
"minimum": 0
},
"before_ms": {
"type": [
"integer",
"null"
],
"format": "uint64",
"minimum": 0
},
"rooms": {
"type": "array",
"items": {
"$ref": "#/$defs/Short"
}
},
"texts": {
"type": "array",
"items": {
"$ref": "#/$defs/Line"
}
},
"top_k": {
"type": "integer",
"format": "uint32",
"minimum": 0
}
},
"required": [
"texts",
"rooms",
"top_k"
]
},
"ClaimOutcome": {
"type": "string",
"enum": [

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

@ -113,6 +113,18 @@
"Sources"
]
},
{
"type": "object",
"properties": {
"Chat": {
"$ref": "#/$defs/ChatHits"
}
},
"additionalProperties": false,
"required": [
"Chat"
]
},
{
"type": "object",
"properties": {
@ -236,6 +248,110 @@
"deletion_line"
]
},
"ChatHit": {
"description": "One chunk of consecutive messages in one room. Every citation is\nrequired: only the room's alias may be absent, because a room may have\nnone. The span and the first and last ids are those of `messages` as\nreturned, after any trimming to the asked span.",
"type": "object",
"properties": {
"first_event_id": {
"type": "string"
},
"last_event_id": {
"type": "string"
},
"messages": {
"type": "array",
"items": {
"$ref": "#/$defs/ChatMessage"
}
},
"room_alias": {
"type": [
"string",
"null"
]
},
"room_id": {
"type": "string"
},
"score": {
"description": "Cosine similarity for one text, the reciprocal-rank sum for several,\nnone in a listing.",
"type": [
"number",
"null"
],
"format": "float"
},
"ts_end_ms": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"ts_start_ms": {
"type": "integer",
"format": "uint64",
"minimum": 0
}
},
"required": [
"room_id",
"first_event_id",
"last_event_id",
"ts_start_ms",
"ts_end_ms",
"messages"
]
},
"ChatHits": {
"description": "The answer to a chat search: oldest first in a listing, nearest first with\ntexts.",
"type": "object",
"properties": {
"hits": {
"type": "array",
"items": {
"$ref": "#/$defs/ChatHit"
}
},
"indexed_under": {
"$ref": "#/$defs/SourceModel"
},
"truncated": {
"description": "A listing had more chunks than `top_k`.",
"type": "boolean"
}
},
"required": [
"indexed_under",
"truncated",
"hits"
]
},
"ChatMessage": {
"description": "One message of a chunk, as the room held it.",
"type": "object",
"properties": {
"body": {
"type": "string"
},
"event_id": {
"type": "string"
},
"sender": {
"description": "The sender's Matrix id.",
"type": "string"
},
"ts_ms": {
"type": "integer",
"format": "uint64",
"minimum": 0
}
},
"required": [
"event_id",
"sender",
"ts_ms",
"body"
]
},
"CitationRole": {
"description": "The referring field of a citation edge, one per leaf field that names\nanother document. The wire spelling is the field's own name, so a client\nreads an edge's role as the field it came from.",
"type": "string",