diff --git a/crates/eureka-state/src/bin/huginn/main.rs b/crates/eureka-state/src/bin/huginn/main.rs index 0b98084..388b76f 100644 --- a/crates/eureka-state/src/bin/huginn/main.rs +++ b/crates/eureka-state/src/bin/huginn/main.rs @@ -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\": , \"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, + /// Messages after it (default 10, at most 50). + #[arg(long)] + after: Option, + /// Print the answer as JSON. + #[arg(long)] + json: bool, + }, } #[derive(Subcommand, Debug)] @@ -321,6 +338,7 @@ fn run(settings: &Settings, command: Command) -> Result { 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, after: Option, json: bool) -> Result { + 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", diff --git a/crates/eureka-state/src/bin/huginn/render.rs b/crates/eureka-state/src/bin/huginn/render.rs index 5c5db1c..09ff943 100644 --- a/crates/eureka-state/src/bin/huginn/render.rs +++ b/crates/eureka-state/src/bin/huginn/render.rs @@ -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; @@ -181,6 +181,27 @@ pub fn chat_hits(found: &ChatHits) -> String { out.join("\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, 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 { diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index 0b81629..cdeb239 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -1310,20 +1310,24 @@ impl huginn_daemon::sources::SourceIndex for Desk { limit: u32, _within: Duration, ) -> anyhow::Result> { - 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> { + fn chat_scroll(&mut self, filter: &huginn_daemon::sources::ChatFilter, order: huginn_daemon::sources::ChatOrder, limit: u32, _within: Duration) -> anyhow::Result> { 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 IndexSink for Desk { self.done.push((ticket, found)); Ok(ticket) } + fn chat_window(&mut self, query: &huginn_mind::ChatWindowQuery) -> Result { + 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) } @@ -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)); +} diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs index 075bfcf..2358f09 100644 --- a/crates/eureka-state/tests/common/mod.rs +++ b/crates/eureka-state/tests/common/mod.rs @@ -97,6 +97,10 @@ impl IndexSink for Inert { Err("this index cannot search".into()) } + fn chat_window(&mut self, _query: &huginn_mind::ChatWindowQuery) -> Result { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &huginn_mind::SourceQuery) -> Result { Err("this index cannot search".into()) } diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index c6bad27..77a968a 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -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 { /// 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; + /// Starts a chat window: the same queue and worker, no embedding. + fn chat_window(&mut self, query: &ChatWindowQuery) -> Result; /// 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> Daemon { 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> Daemon { (_, 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 { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &SourceQuery) -> Result { 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 { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &SourceQuery) -> Result { *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 { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &SourceQuery) -> Result { 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 { + Err("this index cannot search".into()) + } + fn search_sources(&mut self, _query: &SourceQuery) -> Result { Err("this index cannot search".into()) } diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index 7a905fb..8f4db34 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -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> { + fn chat_scroll(&mut self, filter: &ChatFilter, order: ChatOrder, limit: u32, _within: Duration) -> Result> { 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> { self.sources.chat_nearest(vector, filter, limit, within) } - fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result> { - self.sources.chat_scroll(filter, limit, within) + fn chat_scroll(&mut self, filter: &ChatFilter, order: ChatOrder, limit: u32, within: Duration) -> Result> { + self.sources.chat_scroll(filter, order, limit, within) } } diff --git a/crates/huginn-daemon/src/index/mod.rs b/crates/huginn-daemon/src/index/mod.rs index 167972f..067e999 100644 --- a/crates/huginn-daemon/src/index/mod.rs +++ b/crates/huginn-daemon/src/index/mod.rs @@ -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(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 IndexSink for WorkerSink { self.queue_source_ask(SourceAsk::Chat(query.clone())) } + fn chat_window(&mut self, query: &ChatWindowQuery) -> Result { + 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) } diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index 66d08e4..f9fa7a5 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -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> { + fn chat_scroll(&mut self, filter: &ChatFilter, order: ChatOrder, limit: u32, within: Duration) -> Result> { 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,20 @@ 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" } }, + ]) + ); + } } diff --git a/crates/huginn-daemon/src/index/tests.rs b/crates/huginn-daemon/src/index/tests.rs index 1bca83d..b2b61b6 100644 --- a/crates/huginn-daemon/src/index/tests.rs +++ b/crates/huginn-daemon/src/index/tests.rs @@ -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" } })); diff --git a/crates/huginn-daemon/src/sources/mod.rs b/crates/huginn-daemon/src/sources/mod.rs index 7a76ca3..e4cee6a 100644 --- a/crates/huginn-daemon/src/sources/mod.rs +++ b/crates/huginn-daemon/src/sources/mod.rs @@ -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, pub after_ms: Option, pub before_ms: Option, + pub event_id: Option, + pub starts_from_ms: Option, + pub starts_to_ms: Option, +} + +/// 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>; - /// 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>; + fn chat_scroll(&mut self, filter: &ChatFilter, order: ChatOrder, limit: u32, within: Duration) -> Result>; } /// A repo the forge lists as indexable. @@ -651,12 +665,12 @@ pub fn search(embedder: &mut E, index: &mut S, quer /// span and first and last ids recomputed from what is left. Never writes. pub fn search_chat(embedder: &mut E, index: &mut S, query: &ChatQuery, within: Duration) -> Result { 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::>(), truncated) @@ -686,6 +700,95 @@ pub fn search_chat(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(index: &mut S, query: &ChatWindowQuery, within: Duration) -> Result> { + 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()); + 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")?; + let mut held = vec![first_id(&anchor)]; + // Each side walks chunks of the anchor's room only, away from the anchor, + // until it holds more messages than asked (so the cut can tell whether + // more exist), the room ends, or the reads are spent. + let mut walk = |forward: bool, want: usize, mine: usize| -> Result<(Vec>, 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) + }; + let read = index.chat_scroll(&filter, order, WINDOW_CHUNKS_PER_SCROLL, budget.left()?)?; + exhausted = (read.len() as u32) < WINDOW_CHUNKS_PER_SCROLL; + let mut gained = false; + for chunk in read { + let first = first_id(&chunk); + if held.contains(&first) { + continue; + } + held.push(first); + gained = true; + edge = if forward { start_of(&chunk).max(edge) } else { start_of(&chunk).min(edge) }; + count += chunk.messages.len(); + chunks.push(chunk.messages); + } + if exhausted || !gained { + 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: furthest earlier chunk first, the anchor's chunk, then later ones; + // stable by ts_ms, so ties keep chunk order and position. + let ordered = earlier.into_iter().rev().chain(std::iter::once(anchor.messages.clone())).chain(later).flatten(); + let (mut seen, mut messages) = (std::collections::BTreeSet::new(), Vec::new()); + for message in ordered { + if seen.insert(message.event_id.clone()) { + messages.push(message); + } + } + messages.sort_by_key(|message| message.ts_ms); + 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, payload: ChatPayload, filter: &ChatFilter) -> Option { diff --git a/crates/huginn-daemon/src/sources/tests.rs b/crates/huginn-daemon/src/sources/tests.rs index 97a2bb3..adb4646 100644 --- a/crates/huginn-daemon/src/sources/tests.rs +++ b/crates/huginn-daemon/src/sources/tests.rs @@ -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,120 @@ 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 = (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>) -> (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_and_says_more_remain() { + let points: Vec = (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!(around.messages.len() < 51, "six reads of five chunks reach at most 30 earlier messages"); + 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"); +} + +#[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}"); +} diff --git a/crates/huginn-mind/src/lib.rs b/crates/huginn-mind/src/lib.rs index abcd46e..0e41aa9 100644 --- a/crates/huginn-mind/src/lib.rs +++ b/crates/huginn-mind/src/lib.rs @@ -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 diff --git a/crates/huginn-mind/src/wire.rs b/crates/huginn-mind/src/wire.rs index 7952f07..daba414 100644 --- a/crates/huginn-mind/src/wire.rs +++ b/crates/huginn-mind/src/wire.rs @@ -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, } +/// 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, + pub anchor_event_id: String, + pub messages: Vec, + 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), /// 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] diff --git a/docs/chat-retrieval-map.md b/docs/chat-retrieval-map.md index 50b2fc6..f839cd4 100644 --- a/docs/chat-retrieval-map.md +++ b/docs/chat-retrieval-map.md @@ -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 [--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. diff --git a/schemas/cultnet/huginn.mind_request.v1.schema.json b/schemas/cultnet/huginn.mind_request.v1.schema.json index 233da8d..db645c3 100644 --- a/schemas/cultnet/huginn.mind_request.v1.schema.json +++ b/schemas/cultnet/huginn.mind_request.v1.schema.json @@ -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": [ diff --git a/schemas/cultnet/huginn.mind_response.v1.schema.json b/schemas/cultnet/huginn.mind_response.v1.schema.json index 4adbc43..2a2b6d9 100644 --- a/schemas/cultnet/huginn.mind_response.v1.schema.json +++ b/schemas/cultnet/huginn.mind_response.v1.schema.json @@ -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",