From 955b93969c614491ec3ef77e7c3fd02bf67d0346 Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 19:28:43 +0000 Subject: [PATCH 1/2] chat: huginn chat window, a cited event id with its neighbours from the chat index sources::window resolves an event id to its chunk, then walks chunks of the same room both ways (5 per read, at most 6 reads a side), dedupes by event id and cuts to before/after with more_before/more_after. An id the index lacks is a typed SelectionInvalid on event_id naming the three reasons; an id in two chunks is an index fault. No embedder, no Synapse. ChatFilter gains event_id and chunk-start bounds, chat_scroll gains an order, the wire gains ChatWindow request/response, the CLI gains `huginn chat window`. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/src/bin/huginn/main.rs | 37 +++++- crates/eureka-state/src/bin/huginn/render.rs | 23 +++- crates/eureka-state/tests/cli.rs | 53 +++++++- crates/eureka-state/tests/common/mod.rs | 4 + crates/huginn-daemon/src/daemon.rs | 34 ++++- crates/huginn-daemon/src/index/fakes.rs | 13 +- crates/huginn-daemon/src/index/mod.rs | 8 +- crates/huginn-daemon/src/index/qdrant.rs | 33 ++++- crates/huginn-daemon/src/index/tests.rs | 6 +- crates/huginn-daemon/src/sources/mod.rs | 115 ++++++++++++++++- crates/huginn-daemon/src/sources/tests.rs | 117 +++++++++++++++++- crates/huginn-mind/src/lib.rs | 2 +- crates/huginn-mind/src/wire.rs | 82 +++++++++++- docs/chat-retrieval-map.md | 6 + .../huginn.mind_request.v1.schema.json | 49 ++++++++ .../huginn.mind_response.v1.schema.json | 49 ++++++++ 16 files changed, 602 insertions(+), 29 deletions(-) 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", -- 2.47.3 From 0d5efeeaceba7ce74445b6001cc8975d01addfdb Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 19:48:48 +0000 Subject: [PATCH 2/2] chat window: deterministic same-ms order, ties no longer stall the walk, one line per message Order by (ts, chunk start, chunk first event id, position) so the window does not depend on its anchor. Each read asks for the held chunks at the edge in addition, so a run of chunks at one start ms cannot fill a read with held ones; the no-gain early stop is gone. Search and window print a body's line breaks as \n. Tests pin ties, caps, dedupe, the unreached flag and non-chat points. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/src/bin/huginn/render.rs | 15 ++-- crates/eureka-state/tests/cli.rs | 4 +- crates/huginn-daemon/src/index/qdrant.rs | 2 + crates/huginn-daemon/src/sources/mod.rs | 46 +++++++------ crates/huginn-daemon/src/sources/tests.rs | 72 +++++++++++++++++++- 5 files changed, 110 insertions(+), 29 deletions(-) diff --git a/crates/eureka-state/src/bin/huginn/render.rs b/crates/eureka-state/src/bin/huginn/render.rs index 09ff943..3fd7d3d 100644 --- a/crates/eureka-state/src/bin/huginn/render.rs +++ b/crates/eureka-state/src/bin/huginn/render.rs @@ -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 = 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,13 @@ 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 @@ -193,7 +200,7 @@ pub fn chat_window(found: &ChatWindow) -> String { } 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); + 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"); diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index cdeb239..ba446e5 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -2963,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); @@ -3006,7 +3006,7 @@ fn chat_window_prints_the_anchor_marked_and_refuses_what_the_index_lacks() { 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\ + \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); diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index f9fa7a5..3f51c0a 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -562,5 +562,7 @@ mod source_filter_tests { { "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"); } } diff --git a/crates/huginn-daemon/src/sources/mod.rs b/crates/huginn-daemon/src/sources/mod.rs index e4cee6a..963296b 100644 --- a/crates/huginn-daemon/src/sources/mod.rs +++ b/crates/huginn-daemon/src/sources/mod.rs @@ -725,14 +725,12 @@ pub fn window(index: &mut S, query: &ChatWindowQuery, within: Du })); }; 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 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")?; - 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)> { + // 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, 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 { @@ -743,21 +741,23 @@ pub fn window(index: &mut S, query: &ChatWindowQuery, within: Du } 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; + // 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 first = first_id(&chunk); - if held.contains(&first) { + let key = (first_id(&chunk), start_of(&chunk)); + if held.contains(&key) { continue; } - held.push(first); - gained = true; - edge = if forward { start_of(&chunk).max(edge) } else { start_of(&chunk).min(edge) }; + edge = if forward { key.1.max(edge) } else { key.1.min(edge) }; + held.push(key); count += chunk.messages.len(); - chunks.push(chunk.messages); + chunks.push(chunk); } - if exhausted || !gained { + if exhausted { break; } } @@ -767,16 +767,20 @@ pub fn window(index: &mut S, query: &ChatWindowQuery, within: Du }; 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(); + // 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 ordered { + for (.., message) in keyed { 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 { diff --git a/crates/huginn-daemon/src/sources/tests.rs b/crates/huginn-daemon/src/sources/tests.rs index adb4646..fbb364e 100644 --- a/crates/huginn-daemon/src/sources/tests.rs +++ b/crates/huginn-daemon/src/sources/tests.rs @@ -1071,13 +1071,81 @@ fn a_window_at_a_rooms_edge_returns_fewer_and_says_nothing_more_lies_beyond() { } #[test] -fn a_window_stops_after_six_reads_and_says_more_remain() { +fn a_window_stops_after_six_reads_of_five_chunks_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.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 = 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 = (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 = (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] -- 2.47.3