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

chat: huginn chat window (event id with its neighbours, from the chat index) #59

Merged
eureka merged 2 commits from gamecult-chat/chat-window into main 2026-11-11 19:49:05 +00:00
Conversation 0 Commits 2 Files changed 16 +687 -33
16 changed files with 687 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -41,7 +41,7 @@ use huginn_mind::eureka_pipeline::{
Line, PipelineDocument, PipelineRef, Short, Slug, pipeline_key,
};
use huginn_mind::{
BATCH_MAX, ChatQuery, Faculty, HuginnMindRequest, HuginnMindResponse, MindRefusal, PipelineAdmissionBatch, PipelineAdmissionOutcome,
BATCH_MAX, ChatQuery, ChatWindowQuery, Faculty, HuginnMindRequest, HuginnMindResponse, MindRefusal, PipelineAdmissionBatch, PipelineAdmissionOutcome,
PersonaPutOutcome, PipelinePageItems, PipelineProvenance, PipelineSelectionPage, SemanticQuery, SourceQuery,
};
use serde_json::Value;
@ -56,6 +56,7 @@ const DEFAULT_LIMIT: u32 = 50;
const HOLD_CHECK: Duration = Duration::from_secs(10);
/// The neighbours `--near` asks for when `--top-k` is not given.
const DEFAULT_TOP_K: u32 = 10;
const DEFAULT_WINDOW: u32 = 10;
const ADMIT_HELP: &str = "\
The file is a JSON array of 1 to 64 documents, each {\"kind\": <kind>, \"value\": {...}}, or - for stdin; one document object alone is a batch of one. \
@ -98,6 +99,22 @@ enum ChatCommand {
#[arg(long)]
json: bool,
},
/// One indexed message and the messages around it in the same room, by the
/// event id a chat hit cites. Read from the index, so a removed message is
/// refused, not shown.
Window {
/// The event id, beginning `$`.
event_id: String,
/// Messages before it (default 10, at most 50).
#[arg(long)]
before: Option<u32>,
/// Messages after it (default 10, at most 50).
#[arg(long)]
after: Option<u32>,
/// Print the answer as JSON.
#[arg(long)]
json: bool,
},
}
#[derive(Subcommand, Debug)]
@ -321,6 +338,7 @@ fn run(settings: &Settings, command: Command) -> Result<u8, Trouble> {
Command::Chat { command: ChatCommand::Search { texts, rooms, after, before, top_k, json } } => {
chat_search(settings, texts, rooms, after.as_deref(), before.as_deref(), top_k, json)
}
Command::Chat { command: ChatCommand::Window { event_id, before, after, json } } => chat_window(settings, event_id, before, after, json),
Command::Recipe(Recipe::Blocked { spec_id, page }) => {
blocked(settings, &spec_id, page.cursor, page.documents)
}
@ -539,6 +557,22 @@ fn chat_search(
}
}
fn chat_window(settings: &Settings, event_id: String, before: Option<u32>, after: Option<u32>, json: bool) -> Result<u8, Trouble> {
let query = ChatWindowQuery { event_id: Short(event_id), before: before.unwrap_or(DEFAULT_WINDOW), after: after.unwrap_or(DEFAULT_WINDOW) };
match settings.ask(|instance| HuginnMindRequest::ChatWindow { instance: instance.clone(), query })? {
HuginnMindResponse::ChatWindow(found) => {
if json {
say(serde_json::to_string_pretty(&found).map_err(|error| Trouble::Internal(error.to_string()))?);
} else {
say(render::chat_window(&found));
}
Ok(0)
}
HuginnMindResponse::Refused(refusal) => Ok(refused(&refusal)),
other => Err(unexpected(&other)),
}
}
fn show(page: &PipelineSelectionPage) {
match &page.items {
PipelinePageItems::Headers(headers) => {
@ -757,6 +791,7 @@ fn unexpected(answer: &HuginnMindResponse) -> Trouble {
HuginnMindResponse::PersonaPut(_) => "a persona put",
HuginnMindResponse::Sources(_) => "source hits",
HuginnMindResponse::Chat(_) => "chat hits",
HuginnMindResponse::ChatWindow(_) => "a chat window",
HuginnMindResponse::Activity(_) => "repo activity",
HuginnMindResponse::RootsTouched(_) => "roots touched",
HuginnMindResponse::Refused(_) => "a refusal",

36
crates/eureka-state/src/bin/huginn/render.rs
View file

@ -3,7 +3,7 @@
use std::fmt::Write;
use huginn_mind::eureka_pipeline::{PipelineResolution, ResolutionOutcome};
use huginn_mind::{ChatHits, SourceHits, IndexStatus, PersonaStatus, PipelineDocumentSummary, PipelineDocumentView, PipelineStatus, PipelineStatusSummary};
use huginn_mind::{ChatHits, ChatWindow, SourceHits, IndexStatus, PersonaStatus, PipelineDocumentSummary, PipelineDocumentView, PipelineStatus, PipelineStatusSummary};
use serde::Serialize;
use serde_json::Value;
@ -150,8 +150,8 @@ pub fn source_hits(found: &SourceHits) -> String {
/// Chat hits as the CLI prints them. Per hit the organ's own header line,
/// `#alias (!room) first..last time..time score` (the alias and the score only
/// when there are some), then one `time sender event_id body` line per message,
/// the body exactly as the room held it: quoted data under our header, never a
/// message framed as if sent to the reader. A blank line between hits; a
/// the body as the room held it with line breaks written `\n`: quoted data
/// under our header, never a message framed as if sent to the reader. A blank line between hits; a
/// truncated listing ends with the `--after` that continues it.
pub fn chat_hits(found: &ChatHits) -> String {
let mut out: Vec<String> = found
@ -168,7 +168,7 @@ pub fn chat_hits(found: &ChatHits) -> String {
crate::chat::shown(hit.ts_end_ms)
);
for message in &hit.messages {
let _ = writeln!(text, "{} {} {} {}", crate::chat::shown(message.ts_ms), message.sender, message.event_id, message.body);
let _ = writeln!(text, "{} {} {} {}", crate::chat::shown(message.ts_ms), message.sender, message.event_id, one_line(&message.body));
}
text
})
@ -181,6 +181,34 @@ pub fn chat_hits(found: &ChatHits) -> String {
out.join("\n")
}
/// A message body on one line: a line break is written `\n` (and a carriage
/// return `\r`), so a body cannot start a line of its own. `--json` has the
/// body exactly.
fn one_line(body: &str) -> String {
body.replace('\r', "\\r").replace('\n', "\\n")
}
/// A chat window as the CLI prints it: `#alias (!room)` (the id alone with no
/// alias), then one `time sender event_id body` line per message, the anchor
/// marked `>` and the rest indented to match; a line says when messages lie
/// beyond either end.
pub fn chat_window(found: &ChatWindow) -> String {
let room = found.room_alias.as_ref().map_or_else(|| found.room_id.clone(), |alias| format!("{alias} ({})", found.room_id));
let mut out = format!("{room}\n");
if found.more_before {
out.push_str(" (earlier messages exist)\n");
}
for message in &found.messages {
let mark = if message.event_id == found.anchor_event_id { '>' } else { ' ' };
let _ = writeln!(out, "{mark} {} {} {} {}", crate::chat::shown(message.ts_ms), message.sender, message.event_id, one_line(&message.body));
}
if found.more_after {
out.push_str(" (later messages exist)\n");
}
out.pop();
out
}
/// A fact as one token: a string, a number, a flag, a reference's id, or a
/// list of those joined by commas. Anything deeper is not a header fact.
fn scalar(value: &Value) -> Option<String> {

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

@ -1310,20 +1310,24 @@ impl huginn_daemon::sources::SourceIndex for Desk {
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())
Ok(self.chat_scroll(filter, huginn_daemon::sources::ChatOrder::OldestFirst, 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>> {
fn chat_scroll(&mut self, filter: &huginn_daemon::sources::ChatFilter, order: huginn_daemon::sources::ChatOrder, 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())
filter.event_id.as_ref().is_none_or(|id| payload.messages.iter().any(|message| &message.event_id == id))
&& 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);
if order == huginn_daemon::sources::ChatOrder::NewestFirst {
kept.reverse();
}
kept.truncate(limit as usize);
Ok(kept)
}
@ -1357,6 +1361,17 @@ impl<S: MindStore> IndexSink<S> for Desk {
self.done.push((ticket, found));
Ok(ticket)
}
fn chat_window(&mut self, query: &huginn_mind::ChatWindowQuery) -> Result<SearchTicket, String> {
self.tickets += 1;
let ticket = SearchTicket(self.tickets);
let found = match huginn_daemon::sources::window(self, query, Duration::from_secs(5)) {
Ok(Ok(window)) => huginn_daemon::daemon::Found::Window(window),
Ok(Err(refusal)) => huginn_daemon::daemon::Found::Refused(refusal),
Err(error) => huginn_daemon::daemon::Found::Failed(format!("{error:#}")),
};
self.done.push((ticket, found));
Ok(ticket)
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, huginn_daemon::daemon::Found)> {
std::mem::take(&mut self.done)
}
@ -2948,7 +2963,7 @@ fn chat_search_prints_cited_messages_and_sends_its_selectors() {
assert_eq!(
found.out,
"#swarm:gamecult.org (!r:gamecult.org) $e1..$e2 2023-11-14T22:13:20.000Z..2023-11-14T22:14:20.000Z\n\
2023-11-14T22:13:20.000Z @nibu:gamecult.org $e1 ship the cut\nthen rest\n\
2023-11-14T22:13:20.000Z @nibu:gamecult.org $e1 ship the cut\\nthen rest\n\
2023-11-14T22:14:20.000Z @eureka:gamecult.org $e2 done\n\n"
);
let other = at(server.addr, &["chat", "search", "--room", "!elsewhere:gamecult.org"], None);
@ -2974,3 +2989,35 @@ fn chat_search_prints_cited_messages_and_sends_its_selectors() {
assert_eq!(ran.code, 1, "{bad:?} is a refusal by the daemon's door: {}", ran.err);
}
}
/// `huginn chat window` end to end: an unknown or malformed event id is a
/// refusal naming the field; a known one prints the room header and the
/// messages with the anchor marked `>`; the flags reach the request.
#[test]
fn chat_window_prints_the_anchor_marked_and_refuses_what_the_index_lacks() {
let server = source_daemon("d1", "d1");
for (bad, field) in [("$unknown:gamecult.org", "event_id"), ("e1", "event_id")] {
let ran = at(server.addr, &["chat", "window", bad], None);
assert_eq!(ran.code, 1, "{}", ran.err);
assert!(ran.out.starts_with("Refused SelectionInvalid") && ran.out.contains(field), "no window for an id the index lacks: {}", ran.out);
}
let found = at(server.addr, &["chat", "window", "$e2"], None);
assert_eq!(found.code, 0, "{}", found.err);
assert_eq!(
found.out,
"#swarm:gamecult.org (!r:gamecult.org)\n\
\x20 2023-11-14T22:13:20.000Z @nibu:gamecult.org $e1 ship the cut\\nthen rest\n\
> 2023-11-14T22:14:20.000Z @eureka:gamecult.org $e2 done\n"
);
let over = at(server.addr, &["chat", "window", "$e2", "--before", "51"], None);
assert_eq!(over.code, 1, "{}", over.err);
let (spy, seen) = replying(
"chat_window",
HuginnMindResponse::ChatWindow(huginn_mind::ChatWindow { room_id: "!r:x".into(), room_alias: None, anchor_event_id: "$a".into(), messages: vec![], more_before: false, more_after: false }),
);
let ran = at(spy.addr, &["chat", "window", "$a", "--before", "3", "--after", "0"], None);
assert_eq!(ran.code, 0, "{}", ran.err);
let HuginnMindRequest::ChatWindow { query, .. } = seen.lock().unwrap().remove(0) else { panic!("a chat window") };
assert_eq!((query.event_id.0.as_str(), query.before, query.after), ("$a", 3, 0));
}

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

@ -97,6 +97,10 @@ impl<S: MindStore> IndexSink<S> for Inert {
Err("this index cannot search".into())
}
fn chat_window(&mut self, _query: &huginn_mind::ChatWindowQuery) -> 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())
}

34
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::{ChatHits, ChatQuery, Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery, SourceHits, SourceQuery};
use huginn_mind::{ChatHits, ChatQuery, ChatWindow, ChatWindowQuery, 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
@ -37,6 +37,8 @@ pub trait IndexSink<S: MindStore> {
/// 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>;
/// Starts a chat window: the same queue and worker, no embedding.
fn chat_window(&mut self, query: &ChatWindowQuery) -> 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)>;
@ -83,6 +85,9 @@ pub enum Found {
Mind(Hits),
Sources(SourceHits),
Chat(ChatHits),
Window(ChatWindow),
/// The index answered with a typed refusal rather than failing.
Refused(MindRefusal),
Failed(String),
}
@ -219,6 +224,15 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
Err(detail) => Handled::Answered(HuginnMindResponse::Refused(MindRefusal::Unavailable { detail })),
};
}
HuginnMindRequest::ChatWindow { query, .. } => {
if let Err(refusal) = query.check() {
return Handled::Answered(HuginnMindResponse::Refused(refusal));
}
return match self.index.chat_window(&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 {
@ -314,6 +328,8 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
(_, Found::Failed(detail)) => return unavailable(detail),
(None, Found::Sources(hits)) => return HuginnMindResponse::Sources(hits),
(None, Found::Chat(hits)) => return HuginnMindResponse::Chat(hits),
(None, Found::Window(window)) => return HuginnMindResponse::ChatWindow(window),
(None, Found::Refused(refusal)) => return HuginnMindResponse::Refused(refusal),
(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()),
};
@ -599,6 +615,10 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn chat_window(&mut self, _query: &ChatWindowQuery) -> 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())
}
@ -642,6 +662,10 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn chat_window(&mut self, _query: &ChatWindowQuery) -> 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))
@ -768,6 +792,10 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn chat_window(&mut self, _query: &ChatWindowQuery) -> 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())
}
@ -806,6 +834,10 @@ pub(crate) mod tests {
Err("this index cannot search".into())
}
fn chat_window(&mut self, _query: &ChatWindowQuery) -> 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())
}

13
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::{ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta};
use crate::sources::{ChatFilter, ChatOrder, ChatPayload, ChatPoint, ChatRow, DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta};
use super::{CollectionMeta, Described, Embedder, Hit, ModelIdentity, Point, VectorIndex, point_id};
@ -395,10 +395,13 @@ impl SourceIndex for FakeSources {
Ok(rows)
}
fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, _within: Duration) -> Result<Vec<ChatPayload>> {
fn chat_scroll(&mut self, filter: &ChatFilter, order: ChatOrder, 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());
if order == ChatOrder::NewestFirst {
kept.reverse();
}
Ok(kept.into_iter().take(limit as usize).cloned().collect())
}
}
@ -407,7 +410,7 @@ impl SourceIndex for FakeSources {
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),
(Some(start), Some(end)) => filter.event_id.as_ref().is_none_or(|id| payload.messages.iter().any(|message| &message.event_id == id)) && filter.keeps(&payload.room_id, payload.room_alias.as_deref(), start, end),
_ => false,
}
}
@ -437,7 +440,7 @@ impl SourceIndex for FakeIndex {
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)
fn chat_scroll(&mut self, filter: &ChatFilter, order: ChatOrder, limit: u32, within: Duration) -> Result<Vec<ChatPayload>> {
self.sources.chat_scroll(filter, order, limit, within)
}
}

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

@ -107,7 +107,7 @@ use sha2::{Digest, Sha256};
use crate::daemon::{Found, Hits, IndexSink, SearchTicket};
use crate::sources::{self, SourceIndex};
use huginn_mind::{ChatQuery, SourceQuery};
use huginn_mind::{ChatQuery, ChatWindowQuery, SourceQuery};
use ollama::QUERY_INSTRUCTION;
/// The `managed_by` a collection carries when this organ made it.
@ -689,6 +689,7 @@ struct Queued {
enum SourceAsk {
Sources(SourceQuery),
Chat(ChatQuery),
Window(ChatWindowQuery),
}
/// A source or chat search waiting for the worker. It is not the mind's:
@ -833,6 +834,7 @@ fn serve_source_search<E: Embedder, V: VectorIndex + SourceIndex>(projector: &mu
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),
SourceAsk::Window(query) => sources::window(&mut projector.index, &query, within).map(|window| window.map_or_else(Found::Refused, Found::Window)),
}
.unwrap_or_else(|error| Found::Failed(format!("{error:#}")));
lock(shared).sources_found.push((ticket, found));
@ -1121,6 +1123,10 @@ impl<S: MindStore> IndexSink<S> for WorkerSink {
self.queue_source_ask(SourceAsk::Chat(query.clone()))
}
fn chat_window(&mut self, query: &ChatWindowQuery) -> Result<SearchTicket, String> {
self.queue_source_ask(SourceAsk::Window(query.clone()))
}
fn sources_searched(&mut self) -> Vec<(SearchTicket, Found)> {
std::mem::take(&mut lock(&self.shared).sources_found)
}

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

@ -10,7 +10,7 @@ use serde_json::{Value, json};
use huginn_mind::ChatMessage;
use crate::sources::{
COLLECTION, CORPUS_CHAT, CORPUS_FORGE, ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, FRONT_PAGE_FILES, SourceIndex, SourcePayload, SourcePoint, SourceRow,
COLLECTION, CORPUS_CHAT, CORPUS_FORGE, ChatFilter, ChatOrder, ChatPayload, ChatPoint, ChatRow, DescribedSources, FRONT_PAGE_FILES, SourceIndex, SourcePayload, SourcePoint, SourceRow,
SourcesMeta,
};
@ -205,6 +205,15 @@ fn chat_filter(filter: &ChatFilter) -> Value {
if let Some(before) = filter.before_ms {
must.push(json!({ "key": "ts_start_ms", "range": { "lte": before } }));
}
if let Some(from) = filter.starts_from_ms {
must.push(json!({ "key": "ts_start_ms", "range": { "gte": from } }));
}
if let Some(to) = filter.starts_to_ms {
must.push(json!({ "key": "ts_start_ms", "range": { "lte": to } }));
}
if let Some(event_id) = &filter.event_id {
must.push(json!({ "key": "event_ids", "match": { "value": event_id } }));
}
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();
@ -440,13 +449,13 @@ impl SourceIndex for QdrantIndex {
.collect()
}
fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatPayload>> {
fn chat_scroll(&mut self, filter: &ChatFilter, order: ChatOrder, 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" },
"order_by": { "key": "ts_start_ms", "direction": match order { ChatOrder::OldestFirst => "asc", ChatOrder::NewestFirst => "desc" } },
"with_payload": true,
"with_vector": false,
});
@ -521,7 +530,7 @@ mod source_filter_tests {
/// 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) });
let filter = chat_filter(&ChatFilter { rooms: vec!["!r:x".into(), "#a:x".into(), "#b:x".into()], after_ms: Some(5), before_ms: Some(9), ..ChatFilter::default() });
assert_eq!(
filter,
json!({
@ -538,4 +547,22 @@ mod source_filter_tests {
);
assert_eq!(chat_filter(&ChatFilter::default()), json!({ "must": [{ "key": "corpus", "match": { "value": "chat" } }] }));
}
/// An event id is a match on the keyword array, and the start bounds are
/// ranges on the chunk's own start.
#[test]
fn the_chat_filter_matches_an_event_and_bounds_chunk_starts() {
let filter = chat_filter(&ChatFilter { event_id: Some("$e".into()), starts_from_ms: Some(3), starts_to_ms: Some(8), ..ChatFilter::default() });
assert_eq!(
filter["must"],
json!([
{ "key": "corpus", "match": { "value": "chat" } },
{ "key": "ts_start_ms", "range": { "gte": 3 } },
{ "key": "ts_start_ms", "range": { "lte": 8 } },
{ "key": "event_ids", "match": { "value": "$e" } },
])
);
let corpus = json!({ "key": "corpus", "match": { "value": "chat" } });
assert_eq!(filter["must"][0], corpus, "a point with no corpus, or another, does not match: the chat corpus clause is always first");
}
}

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

@ -2175,10 +2175,10 @@ fn source_searches_have_distinct_tickets_share_the_queue_bound_and_abandon_by_ti
/// start first; a nearest search is the filter too.
#[test]
fn qdrant_chat_scroll_and_nearest_send_the_filter_and_the_order() {
use crate::sources::{ChatFilter, SourceIndex};
let filter = ChatFilter { rooms: vec!["!r:x".into()], after_ms: Some(5), before_ms: None };
use crate::sources::{ChatFilter, ChatOrder, SourceIndex};
let filter = ChatFilter { rooms: vec!["!r:x".into()], after_ms: Some(5), before_ms: None, ..ChatFilter::default() };
let scrolling = stub(&[("POST /collections/", 200, r#"{"result":{"points":[]}}"#)]);
QdrantIndex::new(&scrolling.url).chat_scroll(&filter, 7, TIME).unwrap();
QdrantIndex::new(&scrolling.url).chat_scroll(&filter, ChatOrder::OldestFirst, 7, TIME).unwrap();
let scroll: serde_json::Value = serde_json::from_str(&scrolling.seen.lock().unwrap()[0].2).unwrap();
assert_eq!(scroll["order_by"], serde_json::json!({ "key": "ts_start_ms", "direction": "asc" }));
assert_eq!(scroll["filter"]["must"][0], serde_json::json!({ "key": "corpus", "match": { "value": "chat" } }));

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

@ -32,7 +32,7 @@ use std::path::Path;
use std::time::{Duration, Instant};
use anyhow::{Context, Result, bail, ensure};
use huginn_mind::{ChatHit, ChatHits, ChatMessage, ChatQuery, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery};
use huginn_mind::{ChatHit, ChatHits, ChatMessage, ChatQuery, ChatWindow, ChatWindowQuery, MindRefusal, 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};
@ -157,12 +157,24 @@ pub struct ChatRow {
/// 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`].
/// span overlaps [`after_ms`, `before_ms`]. `event_id` keeps the chunk holding
/// that message. `starts_from_ms` and `starts_to_ms` bound a chunk's own start
/// (inclusive), which is what a neighbour read walks by.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ChatFilter {
pub rooms: Vec<String>,
pub after_ms: Option<u64>,
pub before_ms: Option<u64>,
pub event_id: Option<String>,
pub starts_from_ms: Option<u64>,
pub starts_to_ms: Option<u64>,
}
/// Which end of the room's history a chat scroll begins at.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ChatOrder {
OldestFirst,
NewestFirst,
}
impl ChatFilter {
@ -173,6 +185,8 @@ impl ChatFilter {
(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)
&& self.starts_from_ms.is_none_or(|from| start >= from)
&& self.starts_to_ms.is_none_or(|to| start <= to)
}
}
@ -210,9 +224,9 @@ pub trait SourceIndex {
/// 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.
/// Up to `limit` chat chunks `filter` keeps, by span start in `order`.
/// Reads only `corpus=chat` points; the daemon's.
fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatPayload>>;
fn chat_scroll(&mut self, filter: &ChatFilter, order: ChatOrder, limit: u32, within: Duration) -> Result<Vec<ChatPayload>>;
}
/// A repo the forge lists as indexable.
@ -651,12 +665,12 @@ pub fn search<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, quer
/// 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 filter = ChatFilter { rooms: query.rooms.iter().map(|room| room.0.clone()).collect(), after_ms: query.after_ms, before_ms: query.before_ms, ..ChatFilter::default() };
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 mut held = index.chat_scroll(&filter, ChatOrder::OldestFirst, 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)
@ -686,6 +700,99 @@ pub fn search_chat<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S,
})
}
/// Chunks a neighbour read asks for at a time, and the most such reads each way.
const WINDOW_CHUNKS_PER_SCROLL: u32 = 5;
const WINDOW_SCROLLS_MAX: usize = 6;
/// The message a chat window is about and its neighbours, read from the chat
/// index alone: the chunk holding `event_id`, then chunks of that room before
/// and after it until the asked count of messages is held, the room ends or
/// `WINDOW_SCROLLS_MAX` reads are spent. An id the index does not hold is a
/// typed refusal naming `event_id`, never an empty window; an id in two chunks
/// is an index fault. Needs no embedder. Never writes.
pub fn window<S: SourceIndex>(index: &mut S, query: &ChatWindowQuery, within: Duration) -> Result<Result<ChatWindow, MindRefusal>> {
let budget = Budget::new(within);
built_label(index, &budget)?;
let id = query.event_id.0.as_str();
let anchor_filter = ChatFilter { event_id: Some(id.to_owned()), ..ChatFilter::default() };
let mut found = index.chat_scroll(&anchor_filter, ChatOrder::OldestFirst, 2, budget.left()?)?;
ensure!(found.len() < 2, "the chat index holds event {id} in two chunks: an ingest fault, not a window");
let Some(anchor) = found.pop() else {
return Ok(Err(MindRefusal::SelectionInvalid {
field: "event_id".into(),
value: Some(id.to_owned()),
message: "the chat index does not hold this event: it may not be ingested yet, it may have been removed (redacted, or purged by a steward request), or it may be in a room the index does not read (a DM, #users, #members, #patrons, any room the reader has not joined)".into(),
}));
};
let start_of = |chunk: &ChatPayload| chunk.messages.iter().map(|message| message.ts_ms).min().unwrap_or(0);
let first_id = |chunk: &ChatPayload| chunk.messages.first().map(|message| message.event_id.clone()).unwrap_or_default();
let room = ChatFilter { rooms: vec![anchor.room_id.clone()], ..ChatFilter::default() };
let at = anchor.messages.iter().position(|message| message.event_id == id).context("the chat index returned a chunk that does not hold the event it was asked for")?;
// Chunks held so far as (first event id, start): the anchor's, then each walk's.
let mut held = vec![(first_id(&anchor), start_of(&anchor))];
let mut walk = |forward: bool, want: usize, mine: usize| -> Result<(Vec<ChatPayload>, bool)> {
let (mut chunks, mut count, mut edge, mut exhausted) = (Vec::new(), mine, start_of(&anchor), false);
for _ in 0..WINDOW_SCROLLS_MAX {
if count > want {
break;
}
let (filter, order) = if forward {
(ChatFilter { starts_from_ms: Some(edge), ..room.clone() }, ChatOrder::OldestFirst)
} else {
(ChatFilter { starts_to_ms: Some(edge), ..room.clone() }, ChatOrder::NewestFirst)
};
// Held chunks starting at the edge come back first with any others
// that start there; asking for that many extra means a run of
// chunks at one millisecond cannot fill a read with chunks held.
let limit = WINDOW_CHUNKS_PER_SCROLL + held.iter().filter(|(_, start)| *start == edge).count() as u32;
let read = index.chat_scroll(&filter, order, limit, budget.left()?)?;
exhausted = (read.len() as u32) < limit;
for chunk in read {
let key = (first_id(&chunk), start_of(&chunk));
if held.contains(&key) {
continue;
}
edge = if forward { key.1.max(edge) } else { key.1.min(edge) };
held.push(key);
count += chunk.messages.len();
chunks.push(chunk);
}
if exhausted {
break;
}
}
// Stopping short of the asked count without reaching the room's end
// means messages remain that this read did not reach.
Ok((chunks, count <= want && !exhausted))
};
let (earlier, unreached_before) = walk(false, query.before as usize, at)?;
let (later, unreached_after) = walk(true, query.after as usize, anchor.messages.len() - at - 1)?;
// Chronological by (ts, chunk start, chunk first event id, position), so
// the order does not depend on which message anchored the read.
let mut keyed: Vec<(u64, u64, String, usize, ChatMessage)> = Vec::new();
for chunk in earlier.iter().chain(std::iter::once(&anchor)).chain(later.iter()) {
let (start, first) = (start_of(chunk), first_id(chunk));
keyed.extend(chunk.messages.iter().enumerate().map(|(position, message)| (message.ts_ms, start, first.clone(), position, message.clone())));
}
keyed.sort_by(|a, b| (a.0, a.1, &a.2, a.3).cmp(&(b.0, b.1, &b.2, b.3)));
let (mut seen, mut messages) = (std::collections::BTreeSet::new(), Vec::new());
for (.., message) in keyed {
if seen.insert(message.event_id.clone()) {
messages.push(message);
}
}
let at = messages.iter().position(|message| message.event_id == id).context("the anchor fell out of its own window")?;
let (lo, hi) = (at.saturating_sub(query.before as usize), (at + 1 + query.after as usize).min(messages.len()));
Ok(Ok(ChatWindow {
room_id: anchor.room_id,
room_alias: anchor.room_alias,
anchor_event_id: id.to_owned(),
more_before: unreached_before || lo > 0,
more_after: unreached_after || hi < messages.len(),
messages: messages[lo..hi].to_vec(),
}))
}
/// 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> {

185
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::{ChatHits, ChatQuery, SourceQuery};
use huginn_mind::{ChatHits, ChatQuery, ChatWindow, ChatWindowQuery, MindRefusal, SourceQuery};
use super::*;
use crate::index::fakes::{FakeEmbedder, FakeSources};
@ -970,7 +970,188 @@ fn the_span_bounds_are_inclusive_for_chunks_and_for_messages() {
// Both bounds on one message.
assert_eq!(hits(Some(200), Some(200)), owned(&[("$a2", 1)]));
let filter = ChatFilter { rooms: vec![], after_ms: Some(200), before_ms: Some(300) };
let filter = ChatFilter { rooms: vec![], after_ms: Some(200), before_ms: Some(300), ..ChatFilter::default() };
assert!(filter.keeps("!a:x", None, 100, 200) && filter.keeps("!a:x", None, 300, 400));
assert!(!filter.keeps("!a:x", None, 100, 199) && !filter.keeps("!a:x", None, 301, 400));
}
// ---- chat window ----
fn wq(event_id: &str, before: u32, after: u32) -> ChatWindowQuery {
ChatWindowQuery { event_id: Short(event_id.into()), before, after }
}
/// Room `!a`: five two-message chunks, $a0..$a9 at 100..1000. Room `!b`:
/// chunks whose times fall between them ($b0 at 450, $b1 at 550).
fn window_rig() -> Rig {
let mut points: Vec<ChatPoint> = (0..5)
.map(|chunk| {
let (first, second) = (format!("$a{}", chunk * 2), format!("$a{}", chunk * 2 + 1));
let at = chunk as u64 * 200;
chat_point("!a:x", Some("#swarm:x"), 0, 4, &[(&first, at + 100), (&second, at + 200)])
})
.collect();
points.push(chat_point("!b:x", None, 0, 4, &[("$b0", 450), ("$b1", 550)]));
chat_rig(points).0
}
fn window_ids(window: &ChatWindow) -> Vec<&str> {
window.messages.iter().map(|message| message.event_id.as_str()).collect()
}
fn refusal_field(answer: Result<Result<ChatWindow, MindRefusal>>) -> (String, String) {
match answer.unwrap() {
Err(MindRefusal::SelectionInvalid { field, message, .. }) => (field, message),
other => panic!("{other:?}"),
}
}
#[test]
fn an_unknown_event_id_is_refused_naming_the_field_and_the_three_reasons() {
let mut rig = window_rig();
let (field, message) = refusal_field(window(&mut rig.index, &wq("$nope", 3, 3), WITHIN));
assert_eq!(field, "event_id");
for reason in ["not be ingested yet", "removed (redacted, or purged", "a room the index does not read"] {
assert!(message.contains(reason), "{message}");
}
}
#[test]
fn a_redacted_event_is_refused_and_absent_from_its_neighbours_window() {
// The refresh re-chunked the window without $a5: the index holds only $a4.
let mut rig = chat_rig(vec![chat_point("!a:x", None, 0, 4, &[("$a3", 300), ("$a4", 400)]), chat_point("!a:x", None, 0, 4, &[("$a6", 600)])]).0;
assert_eq!(refusal_field(window(&mut rig.index, &wq("$a5", 3, 3), WITHIN)).0, "event_id");
let around = window(&mut rig.index, &wq("$a4", 3, 3), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&around), ["$a3", "$a4", "$a6"]);
}
#[test]
fn an_event_in_two_chunks_is_an_index_fault_not_a_window() {
let mut rig = chat_rig(vec![chat_point("!a:x", None, 0, 4, &[("$a1", 100), ("$a2", 200)]), chat_point("!a:x", None, 0, 4, &[("$a2", 200), ("$a3", 300)])]).0;
let error = window(&mut rig.index, &wq("$a2", 1, 1), WITHIN).unwrap_err().to_string();
assert!(error.contains("$a2") && error.contains("two chunks"), "{error}");
}
#[test]
fn a_window_stays_in_the_anchors_room_though_another_room_is_time_adjacent() {
let mut rig = window_rig();
let around = window(&mut rig.index, &wq("$a4", 4, 4), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&around), ["$a0", "$a1", "$a2", "$a3", "$a4", "$a5", "$a6", "$a7", "$a8"], "no $b between them");
assert_eq!((around.room_id.as_str(), around.room_alias.as_deref()), ("!a:x", Some("#swarm:x")));
let other = window(&mut rig.index, &wq("$b1", 5, 5), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&other), ["$b0", "$b1"], "the other room's window holds none of !a");
assert!(!other.more_before && !other.more_after && other.room_alias.is_none());
}
#[test]
fn a_window_across_chunks_has_exactly_before_and_after_in_order_with_the_anchor_once() {
let mut rig = window_rig();
let around = window(&mut rig.index, &wq("$a4", 3, 3), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&around), ["$a1", "$a2", "$a3", "$a4", "$a5", "$a6", "$a7"]);
assert_eq!(around.anchor_event_id, "$a4");
assert!(around.more_before && around.more_after);
assert!(around.messages.iter().all(|message| !message.event_id.is_empty() && !message.sender.is_empty() && message.ts_ms > 0));
let exact = window(&mut rig.index, &wq("$a4", 4, 5), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&exact).len(), 10);
assert!(!exact.more_before && !exact.more_after, "nothing remains beyond exactly the room's ends");
let alone = window(&mut rig.index, &wq("$a4", 0, 0), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&alone), ["$a4"]);
assert!(alone.more_before && alone.more_after);
}
#[test]
fn a_window_at_a_rooms_edge_returns_fewer_and_says_nothing_more_lies_beyond() {
let mut rig = window_rig();
let start = window(&mut rig.index, &wq("$a0", 5, 1), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&start), ["$a0", "$a1"]);
assert!(!start.more_before && start.more_after);
let end = window(&mut rig.index, &wq("$a9", 1, 5), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&end), ["$a8", "$a9"]);
assert!(end.more_before && !end.more_after);
}
#[test]
fn a_window_stops_after_six_reads_of_five_chunks_and_says_more_remain() {
let points: Vec<ChatPoint> = (0..80).map(|n| chat_point("!a:x", None, 0, 4, &[(&format!("$m{n:02}"), 1000 + n as u64)])).collect();
let mut rig = chat_rig(points).0;
let around = window(&mut rig.index, &wq("$m70", 50, 0), WITHIN).unwrap().unwrap();
assert_eq!(around.messages.len(), 31, "six reads of five new chunks: thirty earlier messages and the anchor");
assert_eq!(around.messages[0].event_id, "$m40");
assert_eq!(around.messages.last().unwrap().event_id, "$m70");
assert!(around.more_before && around.more_after, "the cap is not the room's start, and $m71 follows the anchor");
// Asking for exactly what the six reads reach leaves the same honest flag.
let exact = window(&mut rig.index, &wq("$m70", 30, 0), WITHIN).unwrap().unwrap();
assert_eq!((exact.messages.len(), exact.more_before), (31, true), "thirty held is not yet 'more than asked': the sixth read ended the walk, not the room");
let fewer = window(&mut rig.index, &wq("$m70", 29, 0), WITHIN).unwrap().unwrap();
assert_eq!((fewer.messages.len(), fewer.more_before), (30, true), "the cut leaves $m40 beyond");
}
#[test]
fn a_window_does_not_change_with_the_message_that_anchors_it_when_chunks_share_a_start() {
let chunks = [("$t1", "$t2", 150), ("$u1", "$u2", 160), ("$v1", "$v2", 170)];
let points: Vec<ChatPoint> = chunks.iter().map(|(first, second, at)| chat_point("!a:x", None, 0, 4, &[(first, 100), (second, *at)])).collect();
let mut rig = chat_rig(points).0;
let want = ["$t1", "$u1", "$v1", "$t2", "$u2", "$v2"];
for anchor in want {
let around = window(&mut rig.index, &wq(anchor, 10, 10), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&around), want, "anchored at {anchor}");
assert!(!around.more_before && !around.more_after);
}
}
#[test]
fn a_run_of_chunks_at_one_millisecond_is_all_reached() {
let points: Vec<ChatPoint> = (0..8).map(|n| chat_point("!a:x", None, 0, 4, &[(&format!("$c{n}"), 100)])).collect();
let mut rig = chat_rig(points).0;
for anchor in ["$c0", "$c3", "$c7"] {
let around = window(&mut rig.index, &wq(anchor, 20, 20), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&around), ["$c0", "$c1", "$c2", "$c3", "$c4", "$c5", "$c6", "$c7"], "anchored at {anchor}");
assert!(!around.more_before && !around.more_after, "the room ends in both directions");
}
}
#[test]
fn a_chunk_starting_with_the_anchors_chunk_is_reached_going_forward_alone() {
// The anchor's own chunk already holds a message before it, so no earlier read happens.
let mut rig = chat_rig(vec![chat_point("!a:x", None, 0, 4, &[("$x0", 100), ("$x1", 120)]), chat_point("!a:x", None, 0, 4, &[("$y0", 100), ("$y1", 130)])]).0;
let around = window(&mut rig.index, &wq("$x1", 0, 5), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&around), ["$x1", "$y1"]);
assert!(around.more_before && !around.more_after, "$x0 and $y0 precede; nothing follows $y1");
}
#[test]
fn a_walk_that_holds_exactly_the_asked_count_still_reads_on_to_learn_the_rooms_end() {
let points: Vec<ChatPoint> = (0..10).map(|n| chat_point("!a:x", None, 0, 4, &[(&format!("$m{n}"), 1000 + n as u64)])).collect();
let mut rig = chat_rig(points).0;
let around = window(&mut rig.index, &wq("$m5", 5, 0), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&around), ["$m0", "$m1", "$m2", "$m3", "$m4", "$m5"]);
assert!(!around.more_before, "the room starts at $m0");
}
#[test]
fn an_event_repeated_in_a_neighbour_chunk_appears_once() {
let mut rig = chat_rig(vec![chat_point("!a:x", None, 0, 4, &[("$a1", 100), ("$a2", 200)]), chat_point("!a:x", None, 0, 4, &[("$a2", 200), ("$a3", 300)])]).0;
let around = window(&mut rig.index, &wq("$a1", 0, 5), WITHIN).unwrap().unwrap();
assert_eq!(window_ids(&around), ["$a1", "$a2", "$a3"]);
}
#[test]
fn a_forge_point_is_not_an_event_the_window_can_anchor_on() {
let (mut rig, _) = topic_rig(4);
let mut state = rig.index.state.lock().unwrap();
let mut point = state.points.values().next().unwrap().clone();
point.id = "forge-with-an-event-id".into();
point.payload.text = "$f1 looks like an event id".into();
state.points.insert(point.id.clone(), point);
drop(state);
let (field, _) = refusal_field(window(&mut rig.index, &wq("$f1", 1, 1), WITHIN));
assert_eq!(field, "event_id", "only corpus=chat points are read, as for an unknown id");
}
#[test]
fn a_window_asks_the_index_only_and_needs_no_embedder() {
let source = include_str!("mod.rs");
let signature = &source[source.find("pub fn window").unwrap()..];
let signature = &signature[..signature.find('{').unwrap()];
assert!(!signature.contains("Embedder"), "{signature}");
}

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, CHAT_ROOMS_MAX, ChatHit, ChatHits, ChatMessage, ChatQuery, OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent, RepoMoved, SOURCE_TEXTS_MAX, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery,
ACTIVITY_REPOS_MAX, CHAT_ROOMS_MAX, CHAT_WINDOW_MAX, ChatHit, ChatHits, ChatMessage, ChatQuery, ChatWindow, ChatWindowQuery, EVENT_ID_MAX, 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

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

@ -96,6 +96,9 @@ pub enum HuginnMindRequest {
/// have, by room, by time span and by meaning. Read-only, like
/// `SearchSources`, and queued with it.
SearchChat { instance: Slug, query: ChatQuery },
/// A chat message and its neighbours in the same room, read from the chat
/// index (never from Synapse), by the event id a chat hit cites.
ChatWindow { instance: Slug, query: ChatWindowQuery },
/// 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
@ -364,6 +367,50 @@ pub struct ChatHits {
pub hits: Vec<ChatHit>,
}
/// The most messages a chat window reaches to either side of its anchor.
pub const CHAT_WINDOW_MAX: u32 = 50;
/// The longest event id a chat window takes.
pub const EVENT_ID_MAX: usize = 200;
/// A chat window: the message `event_id` names and up to `before` messages
/// ahead of it and `after` behind it, in its room.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ChatWindowQuery {
pub event_id: Short,
pub before: u32,
pub after: u32,
}
impl ChatWindowQuery {
/// The door. A refusal names the field.
pub fn check(&self) -> Result<(), MindRefusal> {
let refused = |field: &str, value: String, message: &str| MindRefusal::SelectionInvalid { field: field.into(), value: Some(value), message: message.into() };
if !self.event_id.0.starts_with('$') || self.event_id.0.len() > EVENT_ID_MAX {
return Err(refused("event_id", self.event_id.0.clone(), "an event id begins `$` and is at most 200 bytes"));
}
for (field, count) in [("before", self.before), ("after", self.after)] {
if count > CHAT_WINDOW_MAX {
return Err(refused(field, count.to_string(), "a chat window reaches 0 to 50 messages to each side"));
}
}
Ok(())
}
}
/// The answer to a chat window: `messages` chronological with the anchor in
/// them once, all of `room_id`, each carrying the citations a hit's do.
/// `more_before` and `more_after` say messages exist beyond the cut.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct ChatWindow {
pub room_id: String,
pub room_alias: Option<String>,
pub anchor_event_id: String,
pub messages: Vec<ChatMessage>,
pub more_before: bool,
pub more_after: bool,
}
impl HuginnMindRequest {
/// The envelope's `operation`.
pub fn operation(&self) -> &'static str {
@ -376,6 +423,7 @@ impl HuginnMindRequest {
Self::PersonaPut { .. } => "persona_put",
Self::SearchSources { .. } => "search_sources",
Self::SearchChat { .. } => "search_chat",
Self::ChatWindow { .. } => "chat_window",
Self::StewardedActivity { .. } => "stewarded_activity",
Self::RootsTouched { .. } => "roots_touched",
}
@ -392,6 +440,7 @@ impl HuginnMindRequest {
| Self::PersonaPut { instance, .. }
| Self::SearchSources { instance, .. }
| Self::SearchChat { instance, .. }
| Self::ChatWindow { instance, .. }
| Self::StewardedActivity { instance, .. }
| Self::RootsTouched { instance } => Some(instance),
}
@ -412,6 +461,7 @@ pub enum HuginnMindResponse {
PersonaPut(PersonaPutOutcome),
Sources(SourceHits),
Chat(ChatHits),
ChatWindow(ChatWindow),
Activity(RepoActivity),
RootsTouched(Vec<RootTouched>),
/// The answer did not fit one send and is being held for the client to
@ -551,6 +601,7 @@ mod tests {
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::ChatWindow { instance: slug(INSTANCE), query: ChatWindowQuery { event_id: Short("$a".into()), before: 5, after: 0 } },
HuginnMindRequest::RootsTouched { instance: slug(INSTANCE) },
]
}
@ -615,6 +666,14 @@ mod tests {
messages: vec![ChatMessage { event_id: "$a".into(), sender: "@x:gamecult.org".into(), ts_ms: 1, body: "hi".into() }],
}],
}),
HuginnMindResponse::ChatWindow(ChatWindow {
room_id: "!r:gamecult.org".into(),
room_alias: Some("#swarm:gamecult.org".into()),
anchor_event_id: "$a".into(),
messages: vec![ChatMessage { event_id: "$a".into(), sender: "@x:gamecult.org".into(), ts_ms: 1, body: "hi".into() }],
more_before: false,
more_after: true,
}),
HuginnMindResponse::RootsTouched(seeded().roots_touched().unwrap()),
HuginnMindResponse::Deferred(DeferredAnswer {
manifest: cultnet_rs::pack_content("huginn.mind_response", "package", "", "", "", b"body", 4)
@ -630,7 +689,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", "search_chat", "roots_touched"]);
assert_eq!(names, ["whoami", "admit", "view", "query", "persona_get", "persona_put", "search_sources", "search_sources", "search_chat", "chat_window", "roots_touched"]);
for request in requests() {
let declared = request.instance().is_none();
assert_eq!(declared, matches!(request, HuginnMindRequest::Whoami), "{request:?}");
@ -767,6 +826,27 @@ mod tests {
assert_eq!(message, ["event_id", "sender", "ts_ms", "body"]);
}
/// A chat window is refused by field at the door.
#[test]
fn a_chat_window_query_is_refused_by_field() {
let ok = ChatWindowQuery { event_id: Short("$a".into()), before: 50, after: 0 };
assert!(ok.check().is_ok());
let field = |query: ChatWindowQuery| match query.check() {
Err(MindRefusal::SelectionInvalid { field, .. }) => field,
other => panic!("{other:?}"),
};
assert_eq!(field(ChatWindowQuery { before: 51, ..ok.clone() }), "before");
assert_eq!(field(ChatWindowQuery { after: 51, ..ok.clone() }), "after");
assert_eq!(field(ChatWindowQuery { event_id: Short("a".into()), ..ok.clone() }), "event_id");
assert_eq!(field(ChatWindowQuery { event_id: Short(format!("${}", "a".repeat(200))), ..ok.clone() }), "event_id");
assert!(ChatWindowQuery { event_id: Short(format!("${}", "a".repeat(199))), ..ok }.check().is_ok());
let response = serde_json::to_value(schemars::schema_for!(HuginnMindResponse)).unwrap();
let required: Vec<&str> = response["$defs"]["ChatWindow"]["required"].as_array().unwrap().iter().filter_map(|v| v.as_str()).collect();
for field in ["room_id", "anchor_event_id", "messages", "more_before", "more_after"] {
assert!(required.contains(&field), "{field} must be required");
}
}
/// The two files the catalog publishes are the derivation of the two
/// types, byte for byte.
#[test]

6
docs/chat-retrieval-map.md
View file

@ -163,3 +163,9 @@ trimmed start, not by the stored chunk start, and a long chunk trimmed to a span
that crosses rooms can be listed again by the continuation `--after`. Behaviour
is deliberately unchanged here; the event-id window in cut chat-window replaces
the time cursor.
`huginn chat window <EVENT_ID> [--before N] [--after N]` (cut chat-window) reads
the cited message and its neighbours in the same room from this index, no
embedder and no Synapse; an id the index lacks is refused naming `event_id`.
The listing limit above is not changed by it: the cut spec does not assign that
fix, so it stays open.

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

@ -183,6 +183,31 @@
"SearchChat"
]
},
{
"description": "A chat message and its neighbours in the same room, read from the chat\nindex (never from Synapse), by the event id a chat hit cites.",
"type": "object",
"properties": {
"ChatWindow": {
"type": "object",
"properties": {
"instance": {
"$ref": "#/$defs/Slug"
},
"query": {
"$ref": "#/$defs/ChatWindowQuery"
}
},
"required": [
"instance",
"query"
]
}
},
"additionalProperties": false,
"required": [
"ChatWindow"
]
},
{
"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",
@ -341,6 +366,30 @@
"top_k"
]
},
"ChatWindowQuery": {
"description": "A chat window: the message `event_id` names and up to `before` messages\nahead of it and `after` behind it, in its room.",
"type": "object",
"properties": {
"after": {
"type": "integer",
"format": "uint32",
"minimum": 0
},
"before": {
"type": "integer",
"format": "uint32",
"minimum": 0
},
"event_id": {
"$ref": "#/$defs/Short"
}
},
"required": [
"event_id",
"before",
"after"
]
},
"ClaimOutcome": {
"type": "string",
"enum": [

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

@ -125,6 +125,18 @@
"Chat"
]
},
{
"type": "object",
"properties": {
"ChatWindow": {
"$ref": "#/$defs/ChatWindow"
}
},
"additionalProperties": false,
"required": [
"ChatWindow"
]
},
{
"type": "object",
"properties": {
@ -352,6 +364,43 @@
"body"
]
},
"ChatWindow": {
"description": "The answer to a chat window: `messages` chronological with the anchor in\nthem once, all of `room_id`, each carrying the citations a hit's do.\n`more_before` and `more_after` say messages exist beyond the cut.",
"type": "object",
"properties": {
"anchor_event_id": {
"type": "string"
},
"messages": {
"type": "array",
"items": {
"$ref": "#/$defs/ChatMessage"
}
},
"more_after": {
"type": "boolean"
},
"more_before": {
"type": "boolean"
},
"room_alias": {
"type": [
"string",
"null"
]
},
"room_id": {
"type": "string"
}
},
"required": [
"room_id",
"anchor_event_id",
"messages",
"more_before",
"more_after"
]
},
"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",